From 041775435978f4392fc26fad233147cf53c0433e Mon Sep 17 00:00:00 2001 From: Philipp Middendorf Date: Fri, 19 Sep 2025 11:48:13 +0200 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8=20Add=20create/free=20functions=20for?= =?UTF-8?q?=20producer=20and=20consumer?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- ChangeLog | 4 ++++ lib/Asapo/Consumer.hs | 16 +++++++++++++- lib/Asapo/Either/Consumer.hs | 26 +++++++++++++++-------- lib/Asapo/Either/Producer.hs | 41 ++++++++++++++++++++++-------------- lib/Asapo/Producer.hs | 16 +++++++++++++- 5 files changed, 76 insertions(+), 27 deletions(-) diff --git a/ChangeLog b/ChangeLog index 8544f27..700d44a 100644 --- a/ChangeLog +++ b/ChangeLog @@ -1,3 +1,7 @@ +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 diff --git a/lib/Asapo/Consumer.hs b/lib/Asapo/Consumer.hs index 06356c6..edaba1d 100644 --- a/lib/Asapo/Consumer.hs +++ b/lib/Asapo/Consumer.hs @@ -94,7 +94,9 @@ module Asapo.Consumer SourceCredentials (..), NetworkConnectionType (..), - -- * Initialization + -- * Initialization/finalization + createConsumer, + freeConsumer, withConsumer, withGroupId, @@ -296,6 +298,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 withConsumer serverName sourcePath filesystemFlag creds onSuccess = diff --git a/lib/Asapo/Either/Consumer.hs b/lib/Asapo/Either/Consumer.hs index ea2c19e..1044624 100644 --- a/lib/Asapo/Either/Consumer.hs +++ b/lib/Asapo/Either/Consumer.hs @@ -26,6 +26,8 @@ module Asapo.Either.Consumer Error (..), ErrorType (..), withConsumer, + createConsumer, + freeConsumer, retrieveDataForMessageMeta, queryMessages, resendNacs, @@ -170,23 +172,29 @@ withSuccess toCheck onSuccess = do 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 index 238f653..162bf6a 100644 --- a/lib/Asapo/Either/Producer.hs +++ b/lib/Asapo/Either/Producer.hs @@ -35,6 +35,8 @@ module Asapo.Either.Producer setRequestsQueueLimits, checkError, checkErrorWithGivenHandle, + createProducer, + freeProducer, withProducer, enableLocalLog, waitRequestsFinished, @@ -186,21 +188,28 @@ checkError f = do (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 :: @@ -214,14 +223,14 @@ withProducer :: (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 index e3366a4..a45aea5 100644 --- a/lib/Asapo/Producer.hs +++ b/lib/Asapo/Producer.hs @@ -103,7 +103,9 @@ module Asapo.Producer Opcode (..), GenericRequestHeader (..), - -- * Initialization + -- * Initialization/finalization + createProducer, + freeProducer, withProducer, -- * Getters @@ -192,6 +194,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 :: forall a.