diff --git a/ChangeLog b/ChangeLog --- a/ChangeLog +++ b/ChangeLog @@ -1,3 +1,13 @@ +2025-09-12 Philipp Middendorf + + * Add createConsumer and createProducer functions (with free in tandem) + +2025-09-12 Philipp Middendorf + + * Fix memory issue when sending raw data + * Update to latest asapo version + * Make consumer a bit more powerful + 2024-08-16 Philipp Middendorf * Add documentation and high-level bindings diff --git a/hs-asapo.cabal b/hs-asapo.cabal --- a/hs-asapo.cabal +++ b/hs-asapo.cabal @@ -1,6 +1,6 @@ cabal-version: 3.4 name: hs-asapo -version: 0.9.1 +version: 0.9.2 synopsis: Haskell bindings for ASAP:O category: System, FFI, Distributed Computing homepage: https://github.com/pmiddend/hs-asapo diff --git a/lib/Asapo/Consumer.hs b/lib/Asapo/Consumer.hs --- a/lib/Asapo/Consumer.hs +++ b/lib/Asapo/Consumer.hs @@ -94,7 +94,9 @@ SourceCredentials (..), NetworkConnectionType (..), - -- * Initialization + -- * Initialization/finalization + createConsumer, + freeConsumer, withConsumer, withGroupId, @@ -295,6 +297,18 @@ errorTypeToException ErrorUnsupportedClient = throw . UnsupportedClient errorTypeToException ErrorDataNotInCache = throw . DataNotInCache errorTypeToException ErrorUnknownError = throw . UnknownError + +-- | Create a consumer and return a handle. The caller must call 'freeConsumer' after finishing using the handle. See 'withConsumer' for a safer version +createConsumer :: ServerName -> SourcePath -> FilesystemFlag -> SourceCredentials -> IO Consumer +createConsumer serverName sourcePath filesystemFlag sourceCredentials = do + result <- PC.createConsumer serverName sourcePath filesystemFlag sourceCredentials + case result of + Left (Error errorMessage errorType) -> errorTypeToException errorType errorMessage + Right v -> pure v + +-- | Free the consumer handle. This function is only useful in tandem with 'createConsumer' +freeConsumer :: Consumer -> IO () +freeConsumer = PC.freeConsumer -- | Create a consumer and do something with it. This is the main entrypoint into the consumer withConsumer :: forall a. ServerName -> SourcePath -> FilesystemFlag -> SourceCredentials -> (Consumer -> IO a) -> IO a diff --git a/lib/Asapo/Either/Consumer.hs b/lib/Asapo/Either/Consumer.hs --- a/lib/Asapo/Either/Consumer.hs +++ b/lib/Asapo/Either/Consumer.hs @@ -26,6 +26,8 @@ Error (..), ErrorType (..), withConsumer, + createConsumer, + freeConsumer, retrieveDataForMessageMeta, queryMessages, resendNacs, @@ -170,23 +172,29 @@ Left e -> pure (Left e) Right success -> onSuccess success -create :: ServerName -> SourcePath -> FilesystemFlag -> SourceCredentials -> IO (Either Error AsapoConsumerHandle) -create (ServerName serverName) (SourcePath sourcePath) fsFlag creds = +-- | Create a consumer and return a handle. The caller must call 'freeConsumer' after finishing using the handle. See 'withConsumer' for a safer version +createConsumer :: ServerName -> SourcePath -> FilesystemFlag -> SourceCredentials -> IO (Either Error Consumer) +createConsumer (ServerName serverName) (SourcePath sourcePath) fsFlag creds = withCredentials creds \creds' -> withConstText serverName \serverNameC -> withConstText sourcePath \sourcePathC -> - checkError (asapo_create_consumer serverNameC sourcePathC (if fsFlag == WithFilesystem then 1 else 0) creds') + -- first <$> to go into IO, second <$> to go into the Either + (Consumer <$>) <$> checkError (asapo_create_consumer serverNameC sourcePathC (if fsFlag == WithFilesystem then 1 else 0) creds') +-- | Free the consumer handle. This function is only useful in tandem with 'createConsumer' +freeConsumer :: Consumer -> IO () +freeConsumer (Consumer c) = asapo_free_consumer_handle c + -- | Create a consumer and do something with it. This is the main entrypoint into the consumer withConsumer :: forall a. ServerName -> SourcePath -> FilesystemFlag -> SourceCredentials -> (Error -> IO a) -> (Consumer -> IO a) -> IO a -withConsumer serverName sourcePath filesystemFlag creds onError onSuccess = bracket (create serverName sourcePath filesystemFlag creds) freeConsumer handle +withConsumer serverName sourcePath filesystemFlag creds onError onSuccess = bracket (createConsumer serverName sourcePath filesystemFlag creds) freeConsumer' handle where - freeConsumer :: Either Error AsapoConsumerHandle -> IO () - freeConsumer (Right h) = asapo_free_consumer_handle h - freeConsumer _ = pure () - handle :: Either Error AsapoConsumerHandle -> IO a + freeConsumer' :: Either Error Consumer -> IO () + freeConsumer' (Right h) = freeConsumer h + freeConsumer' _ = pure () + handle :: Either Error Consumer -> IO a handle (Left e) = onError e - handle (Right v) = onSuccess (Consumer v) + handle (Right v) = onSuccess v -- | Wrapper around a group ID newtype GroupId = GroupId AsapoStringHandle diff --git a/lib/Asapo/Either/Producer.hs b/lib/Asapo/Either/Producer.hs --- a/lib/Asapo/Either/Producer.hs +++ b/lib/Asapo/Either/Producer.hs @@ -35,6 +35,8 @@ setRequestsQueueLimits, checkError, checkErrorWithGivenHandle, + createProducer, + freeProducer, withProducer, enableLocalLog, waitRequestsFinished, @@ -186,22 +188,29 @@ (errorHandlePtr, result) <- withPtr errorHandle f checkErrorWithGivenHandle errorHandlePtr result -create :: Endpoint -> ProcessingThreads -> RequestHandlerType -> SourceCredentials -> NominalDiffTime -> IO (Either Error AsapoProducerHandle) -create (Endpoint endpoint) (ProcessingThreads processingThreads) handlerType sourceCredentials timeout = do +-- | Create a producer and return a handle. The caller must call 'freeProducer' after finishing using the handle. See 'withProducer' for a safer version +createProducer :: Endpoint -> ProcessingThreads -> RequestHandlerType -> SourceCredentials -> NominalDiffTime -> IO (Either Error Producer) +createProducer (Endpoint endpoint) (ProcessingThreads processingThreads) handlerType sourceCredentials timeout = do withCredentials sourceCredentials \credentials' -> let convertHandlerType TcpHandler = kTcp convertHandlerType FilesystemHandler = kFilesystem in do withText endpoint \endpoint' -> do - checkError - ( asapo_create_producer - endpoint' - (fromIntegral processingThreads) - (convertHandlerType handlerType) - credentials' - (nominalDiffToMillis timeout) - ) + -- first <$> to go into IO, second <$> to go into the Either + (Producer <$>) + <$> checkError + ( asapo_create_producer + endpoint' + (fromIntegral processingThreads) + (convertHandlerType handlerType) + credentials' + (nominalDiffToMillis timeout) + ) +-- | Free the producer handle. This function is only useful in tandem with 'createProducer' +freeProducer :: Producer -> IO () +freeProducer (Producer p) = asapo_free_producer_handle p + -- | Create a producer and do something with it. This is the main entrypoint into the producer withProducer :: forall a. @@ -214,14 +223,14 @@ (Error -> IO a) -> (Producer -> IO a) -> IO a -withProducer endpoint processingThreads handlerType sourceCredentials timeout onError onSuccess = bracket (create endpoint processingThreads handlerType sourceCredentials timeout) freeProducer handle +withProducer endpoint processingThreads handlerType sourceCredentials timeout onError onSuccess = bracket (createProducer endpoint processingThreads handlerType sourceCredentials timeout) freeProducer' handle where - freeProducer :: Either Error AsapoProducerHandle -> IO () - freeProducer (Left _) = pure () - freeProducer (Right producerHandle) = asapo_free_producer_handle producerHandle - handle :: Either Error AsapoProducerHandle -> IO a + freeProducer' :: Either Error Producer -> IO () + freeProducer' (Left _) = pure () + freeProducer' (Right producerHandle) = freeProducer producerHandle + handle :: Either Error Producer -> IO a handle (Left e) = onError e - handle (Right v) = onSuccess (Producer v) + handle (Right v) = onSuccess v withStringHandle :: (AsapoStringHandle -> IO c) -> IO c withStringHandle = bracket asapo_new_string_handle asapo_free_string_handle diff --git a/lib/Asapo/Producer.hs b/lib/Asapo/Producer.hs --- a/lib/Asapo/Producer.hs +++ b/lib/Asapo/Producer.hs @@ -103,7 +103,9 @@ Opcode (..), GenericRequestHeader (..), - -- * Initialization + -- * Initialization/finalization + createProducer, + freeProducer, withProducer, -- * Getters @@ -191,6 +193,18 @@ newtype ProducerException = ProducerException Text deriving (Show) instance Exception ProducerException + +-- | Create a producer and return a handle. The caller must call 'freeProducer' after finishing using the handle. See 'withProducer' for a safer version +createProducer :: Endpoint -> ProcessingThreads -> RequestHandlerType -> SourceCredentials -> NominalDiffTime -> IO Producer +createProducer endpoint processingThreads handlerType sourceCredentials timeout = do + result <- PlainProducer.createProducer endpoint processingThreads handlerType sourceCredentials timeout + case result of + Left (Error errorMessage) -> throw (ProducerException errorMessage) + Right v -> pure v + +-- | Free the producer handle. This function is only useful in tandem with 'createProducer' +freeProducer :: Producer -> IO () +freeProducer = PlainProducer.freeProducer -- | Create a producer and do something with it. This is the main entrypoint into the producer. withProducer ::