Update asapo version

ChangeLog

@@ -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)

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 ];

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

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 ->

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 ->