diff --git a/ChangeLog b/ChangeLog --- a/ChangeLog +++ b/ChangeLog @@ -1,3 +1,7 @@ +2025-11-24 Philipp Middendorf + + * Add getNextAvailable family of functions + 2025-09-12 Philipp Middendorf * Add createConsumer and createProducer functions (with free in tandem) 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.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 diff --git a/lib/Asapo/Consumer.hs b/lib/Asapo/Consumer.hs --- a/lib/Asapo/Consumer.hs +++ b/lib/Asapo/Consumer.hs @@ -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 :: 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 @@ -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 :: 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 @@ -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 ) diff --git a/lib/Asapo/Raw/Consumer.hs b/lib/Asapo/Raw/Consumer.hs --- a/lib/Asapo/Raw/Consumer.hs +++ b/lib/Asapo/Raw/Consumer.hs @@ -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 -> diff --git a/lib/Asapo/Raw/Producer.hsc b/lib/Asapo/Raw/Producer.hsc --- a/lib/Asapo/Raw/Producer.hsc +++ b/lib/Asapo/Raw/Producer.hsc @@ -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