diff --git a/ChangeLog b/ChangeLog index 700d44a..c01d5ba 100644 --- 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/flake.lock b/flake.lock index c22fcb6..5337a32 100644 --- a/flake.lock +++ b/flake.lock @@ -1,15 +1,34 @@ { "nodes": { + "desy-flake": { + "inputs": { + "nixpkgs": "nixpkgs", + "simplon-stub": "simplon-stub" + }, + "locked": { + "lastModified": 1763978305, + "narHash": "sha256-HyacSPDgdFt6NDn5tbtS1a5KoOLKxd/vikEBM6MCsHU=", + "ref": "refs/heads/main", + "rev": "99eb01c54bc44e961fc3fd00e2ee414c9bb3cbae", + "revCount": 13, + "type": "git", + "url": "https://gitlab.desy.de/philipp.middendorf/desy-flake" + }, + "original": { + "type": "git", + "url": "https://gitlab.desy.de/philipp.middendorf/desy-flake" + } + }, "flake-utils": { "inputs": { "systems": "systems" }, "locked": { - "lastModified": 1709126324, - "narHash": "sha256-q6EQdSeUZOG26WelxqkmR7kArjgWCdw5sfJVHPH/7j8=", + "lastModified": 1731533236, + "narHash": "sha256-l0KFg5HjrsfsO/JpG+r7fRrqm12kzFHyUHqHCVpMMbI=", "owner": "numtide", "repo": "flake-utils", - "rev": "d465f4819400de7c8d874d50b982301f28a84605", + "rev": "11707dc2f618dd54ca8739b309ec4fc024de578b", "type": "github" }, "original": { @@ -20,11 +39,42 @@ }, "nixpkgs": { "locked": { - "lastModified": 1757408970, - "narHash": "sha256-aSgK4BLNFFGvDTNKPeB28lVXYqVn8RdyXDNAvgGq+k0=", + "lastModified": 1751741127, + "narHash": "sha256-t75Shs76NgxjZSgvvZZ9qOmz5zuBE8buUaYD28BMTxg=", + "owner": "NixOS", + "repo": "nixpkgs", + "rev": "29e290002bfff26af1db6f64d070698019460302", + "type": "github" + }, + "original": { + "owner": "NixOS", + "ref": "nixos-25.05", + "repo": "nixpkgs", + "type": "github" + } + }, + "nixpkgs_2": { + "locked": { + "lastModified": 1743367904, + "narHash": "sha256-sOos1jZGKmT6xxPvxGQyPTApOunXvScV4lNjBCXd/CI=", "owner": "NixOS", "repo": "nixpkgs", - "rev": "d179d77c139e0a3f5c416477f7747e9d6b7ec315", + "rev": "7ffe0edc685f14b8c635e3d6591b0bbb97365e6c", + "type": "github" + }, + "original": { + "id": "nixpkgs", + "ref": "nixos-24.11", + "type": "indirect" + } + }, + "nixpkgs_3": { + "locked": { + "lastModified": 1763622513, + "narHash": "sha256-1jQnuyu82FpiSxowrF/iFK6Toh9BYprfDqfs4BB+19M=", + "owner": "NixOS", + "repo": "nixpkgs", + "rev": "c58bc7f5459328e4afac201c5c4feb7c818d604b", "type": "github" }, "original": { @@ -35,8 +85,27 @@ }, "root": { "inputs": { + "desy-flake": "desy-flake", "flake-utils": "flake-utils", - "nixpkgs": "nixpkgs" + "nixpkgs": "nixpkgs_3" + } + }, + "simplon-stub": { + "inputs": { + "nixpkgs": "nixpkgs_2" + }, + "locked": { + "lastModified": 1744289673, + "narHash": "sha256-gV0odQHXKnu/P8gtOJR7qaR3QZbEDHxvLqN2Mf0LK6A=", + "owner": "pmiddend", + "repo": "simplon-stub", + "rev": "d93b422e059c85fbd5b9191da69fa32f8c34575d", + "type": "github" + }, + "original": { + "owner": "pmiddend", + "repo": "simplon-stub", + "type": "github" } }, "systems": { diff --git a/flake.nix b/flake.nix index b016ffb..e2be1a4 100644 --- a/flake.nix +++ b/flake.nix @@ -8,15 +8,16 @@ inputs = { nixpkgs.url = "nixpkgs/nixos-25.05"; flake-utils.url = "github:numtide/flake-utils"; + desy-flake.url = "git+https://gitlab.desy.de/philipp.middendorf/desy-flake"; }; - outputs = { self, nixpkgs, flake-utils }: + outputs = { self, nixpkgs, flake-utils, desy-flake }: flake-utils.lib.eachDefaultSystem (system: let pkgs = import nixpkgs { inherit system; - overlays = [ ]; + overlays = [ desy-flake.overlays.default ]; }; haskellPackages = pkgs.haskellPackages.override @@ -25,51 +26,17 @@ }; packageName = "hs-asapo"; - - asapo-core = with pkgs; stdenv.mkDerivation rec { - pname = "asapo"; - - version = "25.03.0"; - - src = fetchurl { - url = "https://gitlab.desy.de/asapo/asapo/-/archive/${version}/asapo-${version}.tar.gz"; - hash = "sha256-DzqjHU4iqunrPTNV22D7FHZTBNzTh1A75qxfc2/VHBE="; - }; - - nativeBuildInputs = [ cmake ]; - - buildInputs = [ - curl - rdkafka - mongoc - cyrus_sasl - # Python is not strictly needed, but the build wants it present. - python3 - ]; - - cmakeFlags = [ - "-DBUILD_PYTHON=OFF" - # This is actually just to let cmake not build the clients. We - # build them ourselves, with Nix methods. - "-DBUILD_CLIENTS_ONLY=ON" - ]; - - # Currently, asapo needs git to evaluate the current branch, which - # doesn't work when you have a tar file as the source. - # patches = [ ./remove-git-references.patch ./fix-kDefaultIngestMode.patch ]; - patches = [ ./remove-git-references.patch ./fix-gcc-14.patch ]; - }; in { packages.${packageName} = haskellPackages.callCabal2nix packageName self { - libasapo-consumer = asapo-core; - libasapo-producer = asapo-core; + libasapo-consumer = pkgs.asapo-libs-devel; + libasapo-producer = pkgs.asapo-libs-devel; }; packages.default = self.packages.${system}.${packageName}; - packages.asapo-core = asapo-core; + packages.asapo-core = pkgs.asapo-libs-devel; defaultPackage = self.packages.${system}.default; @@ -82,7 +49,7 @@ ghcid haskellPackages.hlint haskellPackages.apply-refact - asapo-core + asapo-libs-devel pkg-config ]; inputsFrom = [ self.packages.${system}.hs-asapo.env ]; diff --git a/hs-asapo.cabal b/hs-asapo.cabal index e6ee392..4327647 100644 --- 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 @@ library , 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 index edaba1d..69c152f 100644 --- a/lib/Asapo/Consumer.hs +++ b/lib/Asapo/Consumer.hs @@ -113,6 +113,9 @@ module Asapo.Consumer getNextMessageMetaAndData, getNextMessageMeta, getNextMessageData, + getNextAvailableMessageMetaAndData, + getNextAvailableMessageMeta, + getNextAvailableMessageData, getLastMessageMetaAndData, getLastMessageMeta, getLastMessageData, @@ -457,6 +460,18 @@ getNextMessageMeta consumer streamName groupId = maybeThrow (PC.getNextMessageMe 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 :: Consumer -> diff --git a/lib/Asapo/Either/Consumer.hs b/lib/Asapo/Either/Consumer.hs index 1044624..1caaba7 100644 --- a/lib/Asapo/Either/Consumer.hs +++ b/lib/Asapo/Either/Consumer.hs @@ -43,6 +43,9 @@ module Asapo.Either.Consumer getNextMessageMetaAndData, getNextMessageMeta, getNextMessageData, + getNextAvailableMessageMetaAndData, + getNextAvailableMessageMeta, + getNextAvailableMessageData, getCurrentDatasetCount, getBeamtimeMeta, getCurrentSize, @@ -68,7 +71,7 @@ where 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) @@ -581,6 +584,22 @@ getNextMessageMeta (Consumer consumer) streamName (GroupId groupId) = withMessag 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 :: Consumer -> diff --git a/lib/Asapo/Either/Producer.hs b/lib/Asapo/Either/Producer.hs index 162bf6a..daf4556 100644 --- 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 @@ import Asapo.Raw.Producer AsapoMessageHeaderHandle, AsapoOpcode, AsapoProducerHandle, + AsapoRequestCallback, AsapoRequestCallbackPayloadHandle, asapoLogLevelDebug, asapoLogLevelError, @@ -123,7 +125,7 @@ import Control.Exception (bracket) 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.Maybe (Maybe (Just, Nothing)) 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,23 +488,35 @@ data RequestResponse = RequestResponse 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 :: @@ -529,28 +544,29 @@ send (Producer producer) messageId fileName metadata datasetSubstream datasetSiz (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 @@ sendFile (Producer producer) messageId fileName meta datasetSubstream datasetSiz \messageHeaderHandle -> withConstText fileNameToSend \fileNameToSendC -> withConstText stream \streamC -> do requestCallback <- createRequestCallback (sendRequestCallback callback Nothing) + contextPtr <- new (SendCallbackContext requestCallback) ( fromIntegral <$> ) @@ -593,6 +610,7 @@ sendFile (Producer producer) messageId fileName meta datasetSubstream datasetSiz fileNameToSendC (convertSendFlags transferFlag storageFlag) streamC + (castPtr contextPtr) requestCallback ) @@ -600,7 +618,8 @@ sendFile (Producer producer) messageId fileName meta datasetSubstream datasetSiz 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 @@ sendStreamFinishedFlag (Producer producer) (StreamName stream) (MessageId lastId streamC lastId nextStreamC + (castPtr contextPtr) requestCallback ) @@ -620,7 +640,8 @@ sendBeamtimeMetadata :: Producer -> Metadata -> MetadataIngestMode -> UpsertMode 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 @@ sendBeamtimeMetadata (Producer producer) (Metadata metadata) ingestMode upsertMo UseUpsert -> 1 _ -> 0 ) + (castPtr contextPtr) requestCallback ) @@ -642,7 +664,8 @@ sendStreamMetadata :: Producer -> Metadata -> MetadataIngestMode -> UpsertMode - 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 @@ sendStreamMetadata (Producer producer) (Metadata metadata) ingestMode upsertMode _ -> 0 ) streamC + (castPtr contextPtr) requestCallback ) diff --git a/lib/Asapo/Raw/Consumer.hs b/lib/Asapo/Raw/Consumer.hs index df8a7fd..79ae3a9 100644 --- a/lib/Asapo/Raw/Consumer.hs +++ b/lib/Asapo/Raw/Consumer.hs @@ -66,6 +66,7 @@ module Asapo.Raw.Consumer 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, @@ -421,6 +422,18 @@ foreign import capi "asapo/consumer_c.h asapo_consumer_get_next" 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 -> + Ptr AsapoMessageMetaHandle -> + Ptr AsapoMessageDataHandle -> + -- stream + ConstCString -> + Ptr AsapoErrorHandle -> + IO CInt + foreign import capi "asapo/consumer_c.h asapo_consumer_query_messages" asapo_consumer_query_messages :: AsapoConsumerHandle -> diff --git a/lib/Asapo/Raw/Producer.hsc b/lib/Asapo/Raw/Producer.hsc index 90abdf5..473e418 100644 --- a/lib/Asapo/Raw/Producer.hsc +++ b/lib/Asapo/Raw/Producer.hsc @@ -109,7 +109,11 @@ newtype {-# CTYPE "asapo/producer_c.h" "AsapoMessageHeaderHandle" #-} AsapoMessa 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 @@ foreign import capi "asapo/producer_c.h asapo_producer_send" Word64 -> -- stream ConstCString -> + -- user ptr + Ptr () -> FunPtr AsapoRequestCallback -> Ptr AsapoErrorHandle -> IO CInt @@ -335,6 +341,8 @@ foreign import capi "asapo/producer_c.h asapo_producer_send_file" Word64 -> -- stream ConstCString -> + -- user ptr + Ptr () -> FunPtr AsapoRequestCallback -> Ptr AsapoErrorHandle -> IO CInt @@ -348,6 +356,8 @@ foreign import capi "asapo/producer_c.h asapo_producer_send_stream_finished_flag Word64 -> -- next stream ConstCString -> + -- user ptr + Ptr () -> FunPtr AsapoRequestCallback -> Ptr AsapoErrorHandle -> IO CInt @@ -360,6 +370,8 @@ foreign import capi "asapo/producer_c.h asapo_producer_send_beamtime_metadata" AsapoMetaIngestOp -> -- upsert AsapoBool -> + -- user ptr + Ptr () -> FunPtr AsapoRequestCallback -> Ptr AsapoErrorHandle -> IO CInt @@ -374,6 +386,8 @@ foreign import capi "asapo/producer_c.h asapo_producer_send_stream_metadata" AsapoBool -> -- stream ConstCString -> + -- user data ptr + Ptr () -> FunPtr AsapoRequestCallback -> Ptr AsapoErrorHandle -> IO CInt