✨ Add create/free functions for producer and consumer

ChangeLog

@@ -1,3 +1,7 @@
+2025-09-12  Philipp Middendorf  <philipp.middendorf@desy.de>
+
+	* Add createConsumer and createProducer functions (with free in tandem)
+
 2025-09-12  Philipp Middendorf  <philipp.middendorf@desy.de>
 
 	* Fix memory issue when sending raw data

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 =

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

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

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.