hs-asapo 0.9.2 → 0.10.0

raw patch · 7 files changed

+116/−25 lines, 7 filesdep +derive-storablePVP ok

version bump matches the API change (PVP)

Dependencies added: derive-storable

API changes (from Hackage documentation)

Files

• ChangeLog +4/−0
• hs-asapo.cabal +3/−1
• lib/Asapo/Consumer.hs +15/−0
• lib/Asapo/Either/Consumer.hs +20/−1
• lib/Asapo/Either/Producer.hs +46/−22
• lib/Asapo/Raw/Consumer.hs +13/−0
• lib/Asapo/Raw/Producer.hsc +15/−1

@@ -1,3 +1,7 @@+2025-11-24  Philipp Middendorf  <philipp.middendorf@desy.de>++	* Add getNextAvailable family of functions+ 2025-09-12  Philipp Middendorf  <philipp.middendorf@desy.de>  	* Add createConsumer and createProducer functions (with free in tandem)

@@ -1,6 +1,6 @@ cabal-version:  3.4 name:           hs-asapo-version:        0.9.2+version:        0.10.0 synopsis:       Haskell bindings for ASAP:O category:       System, FFI, Distributed Computing homepage:       https://github.com/pmiddend/hs-asapo@@ -55,6 +55,8 @@     , timerep >= 2.1.0 && < 2.2     -- for the HL interface to send     , bytestring  >= 0.11.5 && < 0.13+    -- for the Storable instance of SendCallback+    , derive-storable >= 0.3.0.0 && < 0.4    pkgconfig-depends: libasapo-consumer                    , libasapo-producer

@@ -113,6 +113,9 @@     getNextMessageMetaAndData,     getNextMessageMeta,     getNextMessageData,+    getNextAvailableMessageMetaAndData,+    getNextAvailableMessageMeta,+    getNextAvailableMessageData,     getLastMessageMetaAndData,     getLastMessageMeta,     getLastMessageData,@@ -456,6 +459,18 @@ -- | Retrieve the next message in the stream and group, only data getNextMessageData :: Consumer -> StreamName -> GroupId -> IO BS.ByteString getNextMessageData consumer streamName groupId = maybeThrow (PC.getNextMessageData consumer streamName groupId)++-- | Retrieve the next available message in the stream and group, with data and metadata+getNextAvailableMessageMetaAndData :: Consumer -> StreamName -> GroupId -> IO (MessageMeta, BS.ByteString)+getNextAvailableMessageMetaAndData consumer streamName groupId = maybeThrow (PC.getNextAvailableMessageMetaAndData consumer streamName groupId)++-- | Retrieve the next available message in the stream and group, only metadata (you can get the data later with 'retrieveDataFromMessageMeta')+getNextAvailableMessageMeta :: Consumer -> StreamName -> GroupId -> IO MessageMeta+getNextAvailableMessageMeta consumer streamName groupId = maybeThrow (PC.getNextAvailableMessageMeta consumer streamName groupId)++-- | Retrieve the next available message in the stream and group, only data+getNextAvailableMessageData :: Consumer -> StreamName -> GroupId -> IO BS.ByteString+getNextAvailableMessageData consumer streamName groupId = maybeThrow (PC.getNextAvailableMessageData consumer streamName groupId)  -- | Query messages, return handles without data queryMessages ::

@@ -43,6 +43,9 @@     getNextMessageMetaAndData,     getNextMessageMeta,     getNextMessageData,+    getNextAvailableMessageMetaAndData,+    getNextAvailableMessageMeta,+    getNextAvailableMessageData,     getCurrentDatasetCount,     getBeamtimeMeta,     getCurrentSize,@@ -68,7 +71,7 @@  import Asapo.Either.Common (MessageId (MessageId), SourceCredentials, StreamInfo, StreamName (StreamName), nominalDiffToMillis, peekConstCStringText, retrieveStreamInfoFromC, stringHandleToText, timespecToUTC, withCStringNToText, withConstText, withCredentials, withPtr) import Asapo.Raw.Common (AsapoErrorHandle, AsapoMessageDataHandle (AsapoMessageDataHandle), AsapoStringHandle, ConstCString, asapo_error_explain, asapo_free_error_handle, asapo_free_stream_infos_handle, asapo_free_string_handle, asapo_is_error, asapo_new_error_handle, asapo_stream_infos_get_item)-import Asapo.Raw.Consumer (AsapoConsumerErrorType, AsapoConsumerHandle, AsapoDataSetHandle, AsapoIdListHandle, AsapoMessageMetaHandle (AsapoMessageMetaHandle), AsapoNetworkConnectionType, AsapoStreamFilter, asapo_consumer_acknowledge, asapo_consumer_current_connection_type, asapo_consumer_delete_stream, asapo_consumer_generate_new_group_id, asapo_consumer_get_beamtime_meta, asapo_consumer_get_by_id, asapo_consumer_get_current_dataset_count, asapo_consumer_get_current_size, asapo_consumer_get_last, asapo_consumer_get_last_dataset, asapo_consumer_get_last_dataset_ingroup, asapo_consumer_get_last_ingroup, asapo_consumer_get_next, asapo_consumer_get_next_dataset, asapo_consumer_get_stream_list, asapo_consumer_get_unacknowledged_messages, asapo_consumer_negative_acknowledge, asapo_consumer_query_messages, asapo_consumer_reset_last_read_marker, asapo_consumer_retrieve_data, asapo_consumer_set_last_read_marker, asapo_consumer_set_resend_nacs, asapo_consumer_set_stream_persistent, asapo_consumer_set_timeout, asapo_create_consumer, asapo_dataset_get_expected_size, asapo_dataset_get_id, asapo_dataset_get_item, asapo_dataset_get_size, asapo_error_get_type, asapo_free_consumer_handle, asapo_free_id_list_handle, asapo_free_message_metas_handle, asapo_id_list_get_item, asapo_id_list_get_size, asapo_message_data_get_as_chars, asapo_message_meta_get_buf_id, asapo_message_meta_get_dataset_substream, asapo_message_meta_get_id, asapo_message_meta_get_metadata, asapo_message_meta_get_name, asapo_message_meta_get_size, asapo_message_meta_get_source, asapo_message_meta_get_timestamp, asapo_message_metas_get_item, asapo_message_metas_get_size, asapo_stream_infos_get_size, kAllStreams, kAsapoTcp, kDataNotInCache, kEndOfStream, kFinishedStreams, kInterruptedTransaction, kLocalIOError, kNoData, kPartialData, kStreamFinished, kUnavailableService, kUndefined, kUnfinishedStreams, kUnsupportedClient, kWrongInput)+import Asapo.Raw.Consumer (AsapoConsumerErrorType, AsapoConsumerHandle, AsapoDataSetHandle, AsapoIdListHandle, AsapoMessageMetaHandle (AsapoMessageMetaHandle), AsapoNetworkConnectionType, AsapoStreamFilter, asapo_consumer_acknowledge, asapo_consumer_current_connection_type, asapo_consumer_delete_stream, asapo_consumer_generate_new_group_id, asapo_consumer_get_beamtime_meta, asapo_consumer_get_by_id, asapo_consumer_get_current_dataset_count, asapo_consumer_get_current_size, asapo_consumer_get_last, asapo_consumer_get_last_dataset, asapo_consumer_get_last_dataset_ingroup, asapo_consumer_get_last_ingroup, asapo_consumer_get_next, asapo_consumer_get_next_available, asapo_consumer_get_next_dataset, asapo_consumer_get_stream_list, asapo_consumer_get_unacknowledged_messages, asapo_consumer_negative_acknowledge, asapo_consumer_query_messages, asapo_consumer_reset_last_read_marker, asapo_consumer_retrieve_data, asapo_consumer_set_last_read_marker, asapo_consumer_set_resend_nacs, asapo_consumer_set_stream_persistent, asapo_consumer_set_timeout, asapo_create_consumer, asapo_dataset_get_expected_size, asapo_dataset_get_id, asapo_dataset_get_item, asapo_dataset_get_size, asapo_error_get_type, asapo_free_consumer_handle, asapo_free_id_list_handle, asapo_free_message_metas_handle, asapo_id_list_get_item, asapo_id_list_get_size, asapo_message_data_get_as_chars, asapo_message_meta_get_buf_id, asapo_message_meta_get_dataset_substream, asapo_message_meta_get_id, asapo_message_meta_get_metadata, asapo_message_meta_get_name, asapo_message_meta_get_size, asapo_message_meta_get_source, asapo_message_meta_get_timestamp, asapo_message_metas_get_item, asapo_message_metas_get_size, asapo_stream_infos_get_size, kAllStreams, kAsapoTcp, kDataNotInCache, kEndOfStream, kFinishedStreams, kInterruptedTransaction, kLocalIOError, kNoData, kPartialData, kStreamFinished, kUnavailableService, kUndefined, kUnfinishedStreams, kUnsupportedClient, kWrongInput) import Asapo.Raw.FreeHandleHack (p_asapo_free_handle) import Control.Applicative (Applicative ((<*>)), pure) import Control.Exception (bracket)@@ -580,6 +583,22 @@ -- | Retrieve the next message in the stream and group, only data getNextMessageData :: Consumer -> StreamName -> GroupId -> IO (Either Error BS.ByteString) getNextMessageData (Consumer consumer) streamName (GroupId groupId) = withMessageHandles streamName (asapo_consumer_get_next consumer groupId) retrieveMessageData++-- | Retrieve the next available message in the stream and group, with data and metadata+getNextAvailableMessageMetaAndData :: Consumer -> StreamName -> GroupId -> IO (Either Error (MessageMeta, BS.ByteString))+getNextAvailableMessageMetaAndData (Consumer consumer) streamName (GroupId groupId) =+  withMessageHandles+    streamName+    (asapo_consumer_get_next_available consumer groupId)+    retrieveMessageMetaAndData++-- | Retrieve the next available message in the stream and group, only metadata (you can get the data later with 'retrieveDataFromMessageMeta')+getNextAvailableMessageMeta :: Consumer -> StreamName -> GroupId -> IO (Either Error MessageMeta)+getNextAvailableMessageMeta (Consumer consumer) streamName (GroupId groupId) = withMessageHandles streamName (asapo_consumer_get_next_available consumer groupId) retrieveMessageMeta++-- | Retrieve the next available message in the stream and group, only data+getNextAvailableMessageData :: Consumer -> StreamName -> GroupId -> IO (Either Error BS.ByteString)+getNextAvailableMessageData (Consumer consumer) streamName (GroupId groupId) = withMessageHandles streamName (asapo_consumer_get_next_available consumer groupId) retrieveMessageData  -- | Query messages, return handles without data queryMessagesHandles ::

@@ -1,4 +1,5 @@ {-# LANGUAGE BlockArguments #-}+{-# LANGUAGE DeriveGeneric #-} {-# LANGUAGE ImportQualifiedPost #-} {-# LANGUAGE ScopedTypeVariables #-} @@ -65,6 +66,7 @@     AsapoMessageHeaderHandle,     AsapoOpcode,     AsapoProducerHandle,+    AsapoRequestCallback,     AsapoRequestCallbackPayloadHandle,     asapoLogLevelDebug,     asapoLogLevelError,@@ -123,7 +125,7 @@ import Data.Bits ((.|.)) import Data.Bool (Bool) import Data.ByteString qualified as BS-import Data.ByteString.Internal qualified as BSI+import Data.ByteString.Unsafe (unsafeUseAsCString) import Data.Either (Either (Left, Right)) import Data.Eq (Eq ((==))) import Data.Foldable (Foldable (elem))@@ -133,11 +135,12 @@ import Data.Ord ((>)) import Data.Text (Text) import Data.Time (NominalDiffTime)-import Data.Word (Word64, Word8)-import Foreign (Storable (peek), alloca, castPtr)+import Data.Word (Word64)+import Foreign (Storable (peek), alloca, castPtr, free, freeHaskellFunPtr, new) import Foreign.C.ConstPtr (ConstPtr (unConstPtr))-import Foreign.ForeignPtr (ForeignPtr, withForeignPtr)-import Foreign.Ptr (Ptr)+import Foreign.Ptr (FunPtr, Ptr)+import Foreign.Storable.Generic (GStorable)+import GHC.Generics (Generic) import System.IO (IO) import Text.Show (Show) import Prelude (fromIntegral)@@ -485,24 +488,36 @@  sendRequestCallback ::   (RequestResponse -> IO ()) ->-  Maybe (ForeignPtr Word8) ->+  Maybe BS.ByteString ->+  -- Maybe (ForeignPtr Word8) ->   Ptr () ->+  Ptr () ->   AsapoRequestCallbackPayloadHandle ->   AsapoErrorHandle ->   IO ()-sendRequestCallback simpleCallback _dataAsForeignPtr _data payloadHandle errorHandle = do+sendRequestCallback simpleCallback _dataAsForeignPtr _data userPtr payloadHandle errorHandle = do   payloadText <- bracket (asapo_request_callback_payload_get_response payloadHandle) asapo_free_string_handle stringHandleToTextUnsafe   originalHeaderCPtr <- asapo_request_callback_payload_get_original_header payloadHandle   originalHeaderC <- peek (unConstPtr originalHeaderCPtr)   originalHeader <- convertRequestHeader originalHeaderC   errorHandle' <- checkErrorWithGivenHandle errorHandle ()+  let sendContextPtr :: Ptr SendCallbackContext+      sendContextPtr = castPtr userPtr+  sendContext <- peek sendContextPtr+  -- First free the closure thingy that Haskell created, then the context it was in+  freeHaskellFunPtr (_sendCallbackContextFunPtr sendContext)+  free sendContextPtr   case errorHandle' of     Left e -> simpleCallback (RequestResponse payloadText originalHeader (Just e))     Right _ -> simpleCallback (RequestResponse payloadText originalHeader Nothing) -toJustForeignPtr :: (a, b, c) -> a-toJustForeignPtr (p, _offset, _length) = p+newtype SendCallbackContext = SendCallbackContext+  { _sendCallbackContextFunPtr :: FunPtr AsapoRequestCallback+  }+  deriving (Generic) +instance GStorable SendCallbackContext+ -- | Send a message containing raw data. Due to newtype and enum usage, all parameter should be self-explanatory send ::   Producer ->@@ -529,28 +544,29 @@     (BS.length data')     \messageHeaderHandle -> do       withConstText stream \streamC -> do-        let word8Ptr :: ForeignPtr Word8-            word8Ptr = toJustForeignPtr (BSI.toForeignPtr data')         requestCallback <-           createRequestCallback             ( sendRequestCallback                 callback-                -- We pass this ptr to the callback to keep the GC from collecting it-                (Just word8Ptr)+                (Just data')             )-        withForeignPtr word8Ptr \dataPtr ->-          ( fromIntegral-              <$>-          )-            <$> checkError+        contextPtr <- new (SendCallbackContext requestCallback)+        -- Technically this is unsafe, because the memory for "data"+        -- might be freed after this call. However, since we pass the+        -- bytestring to the closure, let's hope it's fine.+        unsafeUseAsCString data' \dataPtrAsCString -> do+          result <-+            checkError               ( asapo_producer_send                   producer                   messageHeaderHandle-                  (castPtr dataPtr)+                  (castPtr dataPtrAsCString)                   (convertSendFlags transferFlag storageFlag)                   streamC+                  (castPtr contextPtr)                   requestCallback               )+          pure (fromIntegral <$> result)  -- | Send a message containing a file. Due to newtype and enum usage, all parameter should be self-explanatory sendFile ::@@ -583,6 +599,7 @@     \messageHeaderHandle ->       withConstText fileNameToSend \fileNameToSendC -> withConstText stream \streamC -> do         requestCallback <- createRequestCallback (sendRequestCallback callback Nothing)+        contextPtr <- new (SendCallbackContext requestCallback)         ( fromIntegral             <$>           )@@ -593,6 +610,7 @@                 fileNameToSendC                 (convertSendFlags transferFlag storageFlag)                 streamC+                (castPtr contextPtr)                 requestCallback             ) @@ -600,7 +618,8 @@ sendStreamFinishedFlag :: Producer -> StreamName -> MessageId -> StreamName -> (RequestResponse -> IO ()) -> IO (Either Error Int) sendStreamFinishedFlag (Producer producer) (StreamName stream) (MessageId lastId) (StreamName nextStream) callback = do   requestCallback <- createRequestCallback (sendRequestCallback callback Nothing)-  withConstText stream \streamC -> withConstText nextStream \nextStreamC ->+  withConstText stream \streamC -> withConstText nextStream \nextStreamC -> do+    contextPtr <- new (SendCallbackContext requestCallback)     (fromIntegral <$>)       <$> checkError         ( asapo_producer_send_stream_finished_flag@@ -608,6 +627,7 @@             streamC             lastId             nextStreamC+            (castPtr contextPtr)             requestCallback         ) @@ -620,7 +640,8 @@ sendBeamtimeMetadata (Producer producer) (Metadata metadata) ingestMode upsertMode callback = do   requestCallback <- createRequestCallback (sendRequestCallback callback Nothing)   (fromIntegral <$>)-    <$> withConstText metadata \metadataC ->+    <$> withConstText metadata \metadataC -> do+      contextPtr <- new (SendCallbackContext requestCallback)       checkError         ( asapo_producer_send_beamtime_metadata             producer@@ -634,6 +655,7 @@                 UseUpsert -> 1                 _ -> 0             )+            (castPtr contextPtr)             requestCallback         ) @@ -642,7 +664,8 @@ sendStreamMetadata (Producer producer) (Metadata metadata) ingestMode upsertMode (StreamName stream) callback = do   requestCallback <- createRequestCallback (sendRequestCallback callback Nothing)   (fromIntegral <$>)-    <$> withConstText metadata \metadataC -> withConstText stream \streamC ->+    <$> withConstText metadata \metadataC -> withConstText stream \streamC -> do+      contextPtr <- new (SendCallbackContext requestCallback)       checkError         ( asapo_producer_send_stream_metadata             producer@@ -657,6 +680,7 @@                 _ -> 0             )             streamC+            (castPtr contextPtr)             requestCallback         ) 

@@ -66,6 +66,7 @@     asapo_consumer_get_last,     asapo_consumer_get_last_ingroup,     asapo_consumer_get_next,+    asapo_consumer_get_next_available,     asapo_consumer_query_messages,     asapo_consumer_set_resend_nacs,     asapo_message_data_get_as_chars,@@ -411,6 +412,18 @@  foreign import capi "asapo/consumer_c.h asapo_consumer_get_next"   asapo_consumer_get_next ::+    AsapoConsumerHandle ->+    -- group id+    AsapoStringHandle ->+    Ptr AsapoMessageMetaHandle ->+    Ptr AsapoMessageDataHandle ->+    -- stream+    ConstCString ->+    Ptr AsapoErrorHandle ->+    IO CInt++foreign import capi "asapo/consumer_c.h asapo_consumer_get_next_available"+  asapo_consumer_get_next_available ::     AsapoConsumerHandle ->     -- group id     AsapoStringHandle ->

@@ -109,7 +109,11 @@ asapo_free_message_header_handle :: AsapoMessageHeaderHandle -> IO () asapo_free_message_header_handle (AsapoMessageHeaderHandle ptr) = with ptr \ptr' -> asapo_free_handle ptr' -type AsapoRequestCallback = Ptr () -> AsapoRequestCallbackPayloadHandle -> AsapoErrorHandle -> IO ()+type AsapoRequestCallback =+  -- data ptr+  Ptr () ->+  -- user ptr+  Ptr () -> AsapoRequestCallbackPayloadHandle -> AsapoErrorHandle -> IO ()  foreign import ccall "wrapper" createRequestCallback :: AsapoRequestCallback -> IO (FunPtr AsapoRequestCallback) @@ -321,6 +325,8 @@     Word64 ->     -- stream     ConstCString ->+    -- user ptr+    Ptr () ->     FunPtr AsapoRequestCallback ->     Ptr AsapoErrorHandle ->     IO CInt@@ -335,6 +341,8 @@     Word64 ->     -- stream     ConstCString ->+    -- user ptr+    Ptr () ->     FunPtr AsapoRequestCallback ->     Ptr AsapoErrorHandle ->     IO CInt@@ -348,6 +356,8 @@     Word64 ->     -- next stream     ConstCString ->+    -- user ptr+    Ptr () ->     FunPtr AsapoRequestCallback ->     Ptr AsapoErrorHandle ->     IO CInt@@ -360,6 +370,8 @@     AsapoMetaIngestOp ->     -- upsert     AsapoBool ->+    -- user ptr+    Ptr () ->     FunPtr AsapoRequestCallback ->     Ptr AsapoErrorHandle ->     IO CInt@@ -374,6 +386,8 @@     AsapoBool ->     -- stream     ConstCString ->+    -- user data ptr+    Ptr () ->     FunPtr AsapoRequestCallback ->     Ptr AsapoErrorHandle ->     IO CInt