hs-asapo (empty) → 0.9.0

raw patch · 16 files changed

+3730/−0 lines, 16 filesdep +basedep +bytestringdep +clockbuild-type:Customsetup-changed

Dependencies added: base, bytestring, clock, doctest, hs-asapo, optparse-applicative, text, time, timerep

Files

• ChangeLog +8/−0
• LICENSE +19/−0
• Setup.hs +6/−0
• app/SimpleConsumer.hs +93/−0
• app/SimpleProducer.hs +89/−0
• hs-asapo.cabal +108/−0
• lib/Asapo/Consumer.hs +457/−0
• lib/Asapo/Either/Common.hs +198/−0
• lib/Asapo/Either/Consumer.hs +620/−0
• lib/Asapo/Either/Producer.hs +688/−0
• lib/Asapo/Producer.hs +331/−0
• lib/Asapo/Raw/Common.hs +169/−0
• lib/Asapo/Raw/Consumer.hs +485/−0
• lib/Asapo/Raw/FreeHandleHack.hsc +19/−0
• lib/Asapo/Raw/Producer.hsc +425/−0
• tests/doctests.hs +15/−0

@@ -0,0 +1,8 @@+2024-08-16  Philipp Middendorf  <philipp.middendorf@desy.de>++	* Add documentation and high-level bindings++2024-08-08  Philipp Middendorf  <philipp.middendorf@desy.de>++	* Initial work on 0.9.0, just the raw bindings are here+

@@ -0,0 +1,19 @@+Copyright (c) 2024 Deutsches Elektronen-Synchrotron (DESY)++Permission is hereby granted, free of charge, to any person obtaining a copy+of this software and associated documentation files (the "Software"), to deal+in the Software without restriction, including without limitation the rights+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell+copies of the Software, and to permit persons to whom the Software is+furnished to do so, subject to the following conditions:++The above copyright notice and this permission notice shall be included in all+copies or substantial portions of the Software.++THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE+SOFTWARE.

@@ -0,0 +1,6 @@+module Main where++import Distribution.Extra.Doctest (defaultMainWithDoctests)++main :: IO ()+main = defaultMainWithDoctests "doctests"

@@ -0,0 +1,93 @@+{-# LANGUAGE BlockArguments #-}+{-# LANGUAGE OverloadedStrings #-}++module Main (main) where++import Asapo.Consumer+import Control.Applicative (Applicative ((<*>)))+import Control.Monad (forM_, (=<<), (>>=))+import Data.Bool (Bool (True))+import Data.Either (Either (Left, Right))+import Data.Function (($))+import Data.Functor ((<$>))+import Data.Maybe (Maybe (Nothing), fromMaybe)+import Data.Semigroup (Semigroup ((<>)))+import Data.Text (Text, pack)+import Data.Text.Encoding (decodeUtf8)+import qualified Data.Text.IO as TIO+import Data.Time.Clock (secondsToNominalDiffTime)+import qualified Options.Applicative as Opt+import System.IO (IO)+import Text.Show (Show (show))+import Prelude ()++hstoken :: Token+hstoken = Token "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjk1NzE3MTAyMTYsImp0aSI6Ind0ZmlzdGhpcyIsInN1YiI6ImJ0X2FzYXBvX3Rlc3QiLCJFeHRyYUNsYWltcyI6eyJBY2Nlc3NUeXBlcyI6WyJ3cml0ZSIsIndyaXRlcmF3IiwicmVhZCJdfX0.cz6R_kVf4yh7IJD6bJjDdgTaxPN3txudZx9DE6WaTtk"++data Options = Options+  { optionsServerName :: Text,+    optionsWithFilesystem :: Bool+  }++optionsParser :: Opt.Parser Options+optionsParser =+  Options+    <$> Opt.strOption (Opt.long "server-name")+    <*> Opt.switch (Opt.long "with-filesystem")++main :: IO ()+main = realMain =<< Opt.execParser opts+  where+    opts =+      Opt.info+        (optionsParser Opt.<**> Opt.helper)+        ( Opt.fullDesc+            <> Opt.progDesc "Consume data from asapo"+            <> Opt.header "simple-consumer - a simple message sender"+        )++realMain :: Options -> IO ()+realMain (Options serverName withFilesystem) = do+  withConsumer+    (ServerName serverName)+    (SourcePath "")+    (if withFilesystem then WithFilesystem else WithoutFilesystem)+    ( SourceCredentials+        { sourceType = RawSource,+          instanceId = InstanceId "auto",+          pipelineStep = PipelineStep "ps1",+          beamtime = Beamtime "asapo_test",+          beamline = Beamline "",+          dataSource = DataSource "asapo_source",+          token = hstoken+        }+    )+    \consumer -> do+      TIO.putStrLn "inited consumer"++      TIO.putStrLn "misc: setting timeout"+      setTimeout consumer (secondsToNominalDiffTime 0.5)++      getBeamtimeMeta consumer >>= \meta -> TIO.putStrLn $ "beamtime metadata: " <> (fromMaybe "N/A" meta)++      TIO.putStrLn "listing all available streams:"+      streams <- getStreamList consumer Nothing FilterAllStreams+      forM_ streams \stream -> do+        TIO.putStrLn $ "=> stream info " <> pack (show stream)+        streamSize <- getCurrentSize consumer (streamInfoName stream)+        TIO.putStrLn $ "   stream size: " <> pack (show streamSize)+        datasetCount <- getCurrentDatasetCount consumer (streamInfoName stream) IncludeIncomplete+        TIO.putStrLn $ "   dataset count: " <> pack (show datasetCount)++      -- withGroupId consumer outputError \groupId -> do+      --   onSuccess "getNextMessageMeta" (getNextMessageMeta consumer (streamInfoName stream) groupId) \(messageMetaHandle, messageMeta) -> do+      --     TIO.putStrLn "   got message meta"+      --   onSuccess "getNextMessageMetaAndData" (getNextMessageMetaAndData consumer (streamInfoName stream) groupId) \(messageMetaHandle, messageMeta, messageData) -> do+      --     TIO.putStrLn "   got message"++      (meta, data') <- getMessageMetaAndDataById consumer (StreamName "default") (messageIdFromInt 156)+      TIO.putStrLn $ "meta: " <> pack (show meta)+      TIO.putStrLn $ "data: " <> decodeUtf8 data'++      TIO.putStrLn "misc: resending nacs"+      resendNacs consumer True (secondsToNominalDiffTime 1) 10

@@ -0,0 +1,89 @@+{-# LANGUAGE BlockArguments #-}+{-# LANGUAGE OverloadedStrings #-}++module Main (main) where++import Asapo.Producer+import Control.Applicative (Applicative ((<*>)))+import Control.Monad (void, when, (=<<))+import Data.Bits ((.|.))+import Data.Either (Either (Left, Right))+import Data.Function (($))+import Data.Functor ((<$>))+import Data.Semigroup (Semigroup ((<>)))+import Data.Text (Text, pack)+import Data.Text.Encoding (encodeUtf8)+import qualified Data.Text.IO as TIO+import Data.Time.Clock (secondsToNominalDiffTime)+import Data.Word (Word64)+import Foreign.C.String (newCString, withCString)+import qualified Options.Applicative as Opt+import System.IO (IO)+import Text.Show (Show (show))+import Prelude ()++hstoken :: Token+hstoken = Token "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJleHAiOjk1NzE3MTAyMTYsImp0aSI6Ind0ZmlzdGhpcyIsInN1YiI6ImJ0X2FzYXBvX3Rlc3QiLCJFeHRyYUNsYWltcyI6eyJBY2Nlc3NUeXBlcyI6WyJ3cml0ZSIsIndyaXRlcmF3IiwicmVhZCJdfX0.cz6R_kVf4yh7IJD6bJjDdgTaxPN3txudZx9DE6WaTtk"++data Options = Options+  { optionsMessageId :: Word64,+    optionsMessageContent :: Text,+    optionsEndpoint :: Text,+    optionsStreamName :: Text+  }++optionsParser :: Opt.Parser Options+optionsParser =+  Options+    <$> Opt.option Opt.auto (Opt.long "message-id" <> Opt.help "ID for the message to send (don't re-use, because of duplication messages)")+    <*> Opt.strOption (Opt.long "message-content" <> Opt.help "what to put into the message")+    <*> Opt.strOption (Opt.long "endpoint" <> Opt.help "asapo endpoint")+    <*> Opt.strOption (Opt.long "stream-name")++main :: IO ()+main = realMain =<< Opt.execParser opts+  where+    opts =+      Opt.info+        (optionsParser Opt.<**> Opt.helper)+        ( Opt.fullDesc+            <> Opt.progDesc "Send a string with a specified ID"+            <> Opt.header "simple-producer - a simple message sender"+        )++realMain :: Options -> IO ()+realMain (Options messageId messageContent endpoint streamName) = void $+  withProducer+    (Endpoint endpoint)+    (ProcessingThreads 1)+    TcpHandler+    ( SourceCredentials+        { sourceType = RawSource,+          instanceId = InstanceId "test_instance",+          pipelineStep = PipelineStep "pipeline_step_1",+          beamtime = Beamtime "asapo_test",+          beamline = Beamline "auto",+          dataSource = DataSource "asapo_source",+          token = hstoken+        }+    )+    (secondsToNominalDiffTime 10)+    \producer -> do+      TIO.putStrLn "connected, sending data"+      let responseHandler :: RequestResponse -> IO ()+          responseHandler requestResponse = TIO.putStrLn $ "in response handler, payload " <> responsePayload requestResponse <> ", error " <> pack (show (responseError requestResponse))+      void $+        send+          producer+          (MessageId messageId)+          (FileName $ "raw/" <> streamName <> "/" <> pack (show messageId) <> ".txt")+          (Metadata "{\"test\": 3.0}")+          (DatasetSubstream 0)+          (DatasetSize 0)+          NoAutoId+          (encodeUtf8 messageContent)+          DataAndMetadata+          FilesystemAndDatabase+          (StreamName streamName)+          responseHandler+      waitRequestsFinished producer (secondsToNominalDiffTime 10)

@@ -0,0 +1,108 @@+cabal-version:  3.4+name:           hs-asapo+version:        0.9.0+synopsis:       Haskell bindings for ASAP:O+category:       System, FFI, Distributed Computing+homepage:       https://github.com/pmiddend/hs-asapo+bug-reports:    https://github.com/pmiddend/hs-asapo/issues+author:         Philipp Middendorf+maintainer:     philipp.middendorf@desy.de+copyright:      2024 Philipp Middendorf+license:        MIT+license-file:   LICENSE+build-type:     Custom+extra-source-files: ChangeLog+description:+  Haskell bindings for ASAP:O, a middleware platform for high-performance data analysis. Some general notes about this project:+  +  * @newtype@ and enumerations are used liberally, to make function calls more readable and requiring less documentation (cf. <https://yveskalume.dev/boolean-blindness-dont-represent-state-with-boolean boolean blindness>)+  * For text, such as URLs, identifiers, we assume UTF-8 and use strict [Text]("Data.Text")+  * For data, we copy the data into a strict [ByteString]("Data.ByteString")++  There are two interfaces available: one which does not throw exceptions, but returns a @Either Error a@, and one which throws exceptions. Both expose the same functions. It's yours to decide which one to use.++source-repository head+  type: git+  location: https://github.com/pmiddend/hs-asapo++common warnings+    ghc-options: -Weverything -Wno-missing-safe-haskell-mode -Wno-unsafe -Wno-missing-deriving-strategies -Wno-all-missed-specialisations -Wno-monomorphism-restriction -Wno-safe -Wno-missing-local-signatures -Wno-prepositive-qualified-module -Wno-missing-kind-signatures -Wno-missed-specializations++custom-setup+  setup-depends:+      base < 5+    , cabal-doctest >= 1.0.9 && <1.1++library+  import: warnings+  exposed-modules: Asapo.Raw.Common+                 , Asapo.Raw.Consumer+                 , Asapo.Raw.Producer+                 , Asapo.Raw.FreeHandleHack+                 , Asapo.Either.Producer+                 , Asapo.Either.Consumer+                 , Asapo.Either.Common+                 , Asapo.Producer+                 , Asapo.Consumer+  build-depends:+      base >=4.7 && <5+    , text >= 2.0.2 && < 2.1+    -- for timespec data type (is Storable)+    , clock >= 0.8.4 && < 0.9+    -- for higher-level UTCTime/LocalTime+    , time >= 1.12.2 && < 1.13+    -- for RFC3339+    , timerep >= 2.1.0 && < 2.2+    -- for the HL interface to send+    , bytestring  >= 0.11.5 && < 0.12++  pkgconfig-depends: libasapo-consumer+                   , libasapo-producer+  hs-source-dirs:    lib++-- Marked as "benchmark" so the dependencies on hackage are accurate+benchmark simple-producer+  type: exitcode-stdio-1.0+  main-is: SimpleProducer.hs+  +  ghc-options: -threaded -rtsopts -with-rtsopts=-N+  hs-source-dirs:+      app+  build-depends:+      base >=4.7 && <5+    , hs-asapo+    , text >= 2.0.2 && < 2.1+    , time >= 1.12.2 && < 1.13+    , optparse-applicative >= 0.18.1 && < 0.19+    +-- Marked as "benchmark" so the dependencies on hackage are accurate+benchmark simple-consumer+  type: exitcode-stdio-1.0+  main-is: SimpleConsumer.hs+  +  ghc-options: -threaded -rtsopts -with-rtsopts=-N+  hs-source-dirs:+      app+  build-depends:+      base >=4.7 && <5+    , hs-asapo+    , text >= 2.0.2 && < 2.1+    , time >= 1.12.2 && < 1.13+    , optparse-applicative+    +-- copied from https://github.com/ulidtko/cabal-doctest+test-suite doctests+  type:             exitcode-stdio-1.0+  main-is:          doctests.hs+  build-depends:+      base >=4.7 && <5+    , doctest >=0.15   && <1+    , clock >= 0.8.4 && < 0.9+    , timerep >= 2.1.0 && < 2.2+    , time >= 1.12.2 && < 1.13+    , bytestring  >= 0.11.5 && < 0.12+    , text >= 2.0.2 && < 2.1+    , hs-asapo++  ghc-options:      -Wall -threaded+  hs-source-dirs:   tests

@@ -0,0 +1,457 @@+{-# LANGUAGE ExistentialQuantification #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# OPTIONS_GHC -Wno-missing-import-lists #-}++-- |+-- Description : High-level interface for all consumer-related functions, using exceptions instead of @Either@+--+-- To implement an ASAP:O consumer, you should only need this interface.+-- It exposes no memory-management functions (like free) or pointers, and+-- is thus safe to use.+--+-- = Simple Example+--+-- Here's some code for a simple consumer that connects, outputs the available streams, and then gets a specific message by ID with metadata and data:+--+-- >>> :seti -XOverloadedStrings+-- >>> :{+--  module Main where+--  import Asapo.Consumer+--  import Prelude hiding (putStrLn)+--  import Data.Maybe(fromMaybe)+--  import Control.Monad(forM_)+--  import Data.Text(pack)+--  import Data.Text.IO(putStrLn)+--  import Data.Text.Encoding(decodeUtf8)+--  import qualified Data.ByteString as BS+--  main :: IO ()+--  main =+--    withConsumer+--      (ServerName "localhost:8040")+--      (SourcePath "")+--      WithoutFilesystem+--      ( SourceCredentials+--        { sourceType = RawSource,+--          instanceId = InstanceId "auto",+--          pipelineStep = PipelineStep "ps1",+--          beamtime = Beamtime "asapo_test",+--          beamline = Beamline "",+--          dataSource = DataSource "asapo_source",+--          token = Token "token-please-change"+--        }+--      ) $ \consumer -> do+--        beamtimeMeta <- getBeamtimeMeta consumer+--        putStrLn $ "beamtime metadata: " <> (fromMaybe "N/A" beamtimeMeta)+--        streams <- getStreamList consumer Nothing FilterAllStreams+--        forM_ streams $ \stream -> do+--          putStrLn $ "=> stream info " <> pack (show stream)+--          streamSize <- getCurrentSize consumer (streamInfoName stream)+--          putStrLn $ "   stream size: " <> pack (show streamSize)+--          datasetCount <- getCurrentDatasetCount consumer (streamInfoName stream) IncludeIncomplete+--          putStrLn $ "   dataset count: " <> pack (show datasetCount)+--        metaAndData <- getMessageMetaAndDataById consumer (StreamName "default") (messageIdFromInt 1337)+--        putStrLn $ "meta: " <> pack (show (fst metaAndData))+--        putStrLn $ "data: " <> decodeUtf8 (snd metaAndData)+-- :}+module Asapo.Consumer+  ( -- * Error types+    SomeConsumerException,+    NoData,+    EndOfStream,+    StreamFinished,+    UnavailableService,+    InterruptedTransaction,+    LocalIOError,+    WrongInput,+    PartialData,+    UnsupportedClient,+    DataNotInCache,+    UnknownError,++    -- * Types+    Consumer,+    Dataset (..),+    MessageMetaHandle,+    DeleteFlag (..),+    ErrorOnNotExistFlag (..),+    ErrorType (ErrorNoData),+    FilesystemFlag (..),+    PipelineStep (..),+    Beamtime (Beamtime),+    DataSource (DataSource),+    Beamline (Beamline),+    StreamInfo (..),+    GroupId,+    IncludeIncompleteFlag (..),+    MessageMeta (..),+    ServerName (..),+    SourcePath (..),+    SourceType (..),+    StreamName (..),+    InstanceId (..),+    Token (..),+    StreamFilter (..),+    SourceCredentials (..),+    NetworkConnectionType (..),++    -- * Initialization+    withConsumer,+    withGroupId,++    -- * Getters+    getCurrentSize,+    getCurrentDatasetCount,+    getBeamtimeMeta,+    getNextDataset,+    getLastDataset,+    getLastDatasetInGroup,+    getMessageMetaAndDataById,+    getMessageMetaById,+    getMessageDataById,+    getNextMessageMetaAndData,+    getNextMessageMeta,+    getNextMessageData,+    getLastMessageMetaAndData,+    getLastMessageMeta,+    getLastMessageData,+    getLastInGroupMessageMetaAndData,+    getLastInGroupMessageMeta,+    getLastInGroupMessageData,+    getUnacknowledgedMessages,+    getCurrentConnectionType,+    getStreamList,+    messageIdFromInt,+    queryMessages,+    retrieveDataForMessageMeta,++    -- * Modifiers+    resetLastReadMarker,+    setTimeout,+    setLastReadMarker,+    setStreamPersistent,+    acknowledge,+    negativeAcknowledge,+    deleteStream,+    resendNacs,+  )+where++import Asapo.Either.Common+  ( Beamline (Beamline),+    Beamtime (Beamtime),+    DataSource (DataSource),+    InstanceId (..),+    MessageId,+    PipelineStep (..),+    SourceCredentials (..),+    SourceType (..),+    StreamInfo (..),+    StreamName (..),+    Token (..),+    messageIdFromInt,+  )+import Asapo.Either.Consumer+  ( Consumer,+    Dataset (..),+    DeleteFlag (..),+    Error (Error),+    ErrorOnNotExistFlag (..),+    ErrorType (..),+    FilesystemFlag (..),+    GroupId,+    IncludeIncompleteFlag (..),+    MessageMeta (..),+    MessageMetaHandle,+    NetworkConnectionType (..),+    ServerName (..),+    SourcePath (..),+    StreamFilter (..),+    getCurrentConnectionType,+    resendNacs,+    setTimeout,+  )+import qualified Asapo.Either.Consumer as PC+import Control.Applicative (pure)+import Control.Exception (Exception (fromException, toException), SomeException, throw)+import Control.Monad (Monad)+import qualified Data.ByteString as BS+import Data.Either (Either (Left, Right))+import Data.Function ((.))+import Data.Int (Int)+import Data.Maybe (Maybe)+import Data.Text (Text)+import Data.Time (NominalDiffTime)+import Data.Typeable (cast)+import Data.Word (Word64)+import System.IO (IO)+import Text.Show (Show, show)+import Prelude ()++-- $setup+-- >>> :seti -XOverloadedStrings+-- >>> import Control.Exception(catch)+-- >>> import Prelude(undefined, (<>), error)+-- >>> import Data.Text.IO(putStrLn)+-- >>> import Data.Text(pack)+-- >>> let consumer = undefined++-- | Parent class for all consumer-related exceptions. This makes catchall possible, as in...+--+-- @+-- setStreamPersistent consumer (StreamName "default")+--   `catch` (\e -> error ("Caught " <> (show (e :: SomeConsumerException))))+-- @+data SomeConsumerException+  = forall e. (Exception e) => SomeConsumerException e++instance Show SomeConsumerException where+  show (SomeConsumerException e) = show e++instance Exception SomeConsumerException++consumerExceptionToException :: (Exception e) => e -> SomeException+consumerExceptionToException = toException . SomeConsumerException++consumerExceptionFromException :: (Exception e) => SomeException -> Maybe e+consumerExceptionFromException x = do+  SomeConsumerException a <- fromException x+  cast a++newtype NoData = NoData Text deriving (Show)++instance Exception NoData where+  toException = consumerExceptionToException+  fromException = consumerExceptionFromException++newtype EndOfStream = EndOfStream Text deriving (Show)++instance Exception EndOfStream where+  toException = consumerExceptionToException+  fromException = consumerExceptionFromException++newtype StreamFinished = StreamFinished Text deriving (Show)++instance Exception StreamFinished where+  toException = consumerExceptionToException+  fromException = consumerExceptionFromException++newtype UnavailableService = UnavailableService Text deriving (Show)++instance Exception UnavailableService where+  toException = consumerExceptionToException+  fromException = consumerExceptionFromException++newtype InterruptedTransaction = InterruptedTransaction Text deriving (Show)++instance Exception InterruptedTransaction where+  toException = consumerExceptionToException+  fromException = consumerExceptionFromException++newtype LocalIOError = LocalIOError Text deriving (Show)++instance Exception LocalIOError where+  toException = consumerExceptionToException+  fromException = consumerExceptionFromException++newtype WrongInput = WrongInput Text deriving (Show)++instance Exception WrongInput where+  toException = consumerExceptionToException+  fromException = consumerExceptionFromException++newtype PartialData = PartialData Text deriving (Show)++instance Exception PartialData where+  toException = consumerExceptionToException+  fromException = consumerExceptionFromException++newtype UnsupportedClient = UnsupportedClient Text deriving (Show)++instance Exception UnsupportedClient where+  toException = consumerExceptionToException+  fromException = consumerExceptionFromException++newtype DataNotInCache = DataNotInCache Text deriving (Show)++instance Exception DataNotInCache where+  toException = consumerExceptionToException+  fromException = consumerExceptionFromException++newtype UnknownError = UnknownError Text deriving (Show)++instance Exception UnknownError where+  toException = consumerExceptionToException+  fromException = consumerExceptionFromException++errorTypeToException :: ErrorType -> Text -> a+errorTypeToException ErrorNoData = throw . NoData+errorTypeToException ErrorEndOfStream = throw . EndOfStream+errorTypeToException ErrorStreamFinished = throw . StreamFinished+errorTypeToException ErrorUnavailableService = throw . UnavailableService+errorTypeToException ErrorInterruptedTransaction = throw . InterruptedTransaction+errorTypeToException ErrorLocalIOError = throw . LocalIOError+errorTypeToException ErrorWrongInput = throw . WrongInput+errorTypeToException ErrorPartialData = throw . PartialData+errorTypeToException ErrorUnsupportedClient = throw . UnsupportedClient+errorTypeToException ErrorDataNotInCache = throw . DataNotInCache+errorTypeToException ErrorUnknownError = throw . UnknownError++-- | 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 =+  let onError (Error errorMessage errorType) = errorTypeToException errorType errorMessage+   in PC.withConsumer serverName sourcePath filesystemFlag creds onError onSuccess++-- | Allocate a group ID and call a callback+withGroupId :: forall a. Consumer -> (GroupId -> IO a) -> IO a+withGroupId consumer onSuccess =+  let onError (Error errorMessage errorType) = errorTypeToException errorType errorMessage+   in PC.withGroupId consumer onError onSuccess++maybeThrow :: (Monad m) => m (Either Error b) -> m b+maybeThrow f = do+  result <- f+  case result of+    Left (Error errorMessage errorType) -> errorTypeToException errorType errorMessage+    Right v -> pure v++-- | Reset the last read marker for the stream+resetLastReadMarker :: Consumer -> GroupId -> StreamName -> IO Int+resetLastReadMarker consumer groupId streamName = maybeThrow (PC.resetLastReadMarker consumer groupId streamName)++-- | Set the last read marker for the stream+setLastReadMarker :: Consumer -> GroupId -> StreamName -> MessageId -> IO Int+setLastReadMarker consumer groupId streamName value = maybeThrow (PC.setLastReadMarker consumer groupId streamName value)++-- | Acknowledge a specific message+acknowledge :: Consumer -> GroupId -> StreamName -> MessageId -> IO Int+acknowledge consumer groupId streamName messageId = maybeThrow (PC.acknowledge consumer groupId streamName messageId)++-- | Negatively acknowledge a specific message+negativeAcknowledge ::+  Consumer ->+  GroupId ->+  StreamName ->+  MessageId ->+  -- | delay+  NominalDiffTime ->+  IO Int+negativeAcknowledge consumer groupId streamName messageId delay = maybeThrow (PC.negativeAcknowledge consumer groupId streamName messageId delay)++-- | Get a list of all unacknowledged message IDs in a range+getUnacknowledgedMessages :: Consumer -> GroupId -> StreamName -> (MessageId, MessageId) -> IO [MessageId]+getUnacknowledgedMessages consumer groupId streamName (from, to) = maybeThrow (PC.getUnacknowledgedMessages consumer groupId streamName (from, to))++-- | Retrieve the list of streams with metadata+getStreamList :: Consumer -> Maybe StreamName -> StreamFilter -> IO [StreamInfo]+getStreamList consumer streamName filter = maybeThrow (PC.getStreamList consumer streamName filter)++-- | Delete a given stream+deleteStream :: Consumer -> StreamName -> DeleteFlag -> ErrorOnNotExistFlag -> IO Int+deleteStream consumer streamName deleteFlag errorOnNotExistFlag = maybeThrow (PC.deleteStream consumer streamName deleteFlag errorOnNotExistFlag)++-- | Set a stream persistent+setStreamPersistent :: Consumer -> StreamName -> IO Int+setStreamPersistent consumer streamName = maybeThrow (PC.setStreamPersistent consumer streamName)++-- | Get the current size (number of messages) of the stream+getCurrentSize :: Consumer -> StreamName -> IO Int+getCurrentSize consumer streamName = maybeThrow (PC.getCurrentSize consumer streamName)++-- | Get number of datasets in stream+getCurrentDatasetCount :: Consumer -> StreamName -> IncludeIncompleteFlag -> IO Int+getCurrentDatasetCount consumer streamName inludeIncomplete = maybeThrow (PC.getCurrentDatasetCount consumer streamName inludeIncomplete)++-- | Get beamtime metadata (which can be not set, in which case @Nothing@ is returned)+getBeamtimeMeta :: Consumer -> IO (Maybe Text)+getBeamtimeMeta consumer = maybeThrow (PC.getBeamtimeMeta consumer)++-- | Get the next dataset for a stream+getNextDataset ::+  Consumer ->+  GroupId ->+  -- | minimum size+  Word64 ->+  StreamName ->+  IO Dataset+getNextDataset consumer groupId minSize streamName = maybeThrow (PC.getNextDataset consumer groupId minSize streamName)++-- | Get the last dataset in the stream+getLastDataset ::+  Consumer ->+  -- | minimum size+  Word64 ->+  StreamName ->+  IO Dataset+getLastDataset consumer minSize streamName = maybeThrow (PC.getLastDataset consumer minSize streamName)++-- | Get the last data ste in the given group+getLastDatasetInGroup ::+  Consumer ->+  GroupId ->+  -- | minimum size+  Word64 ->+  StreamName ->+  IO Dataset+getLastDatasetInGroup consumer groupId minSize streamName = maybeThrow (PC.getLastDatasetInGroup consumer groupId minSize streamName)++-- | Given a message ID, retrieve both metadata and data+getMessageMetaAndDataById :: Consumer -> StreamName -> MessageId -> IO (MessageMeta, BS.ByteString)+getMessageMetaAndDataById consumer streamName messageId = maybeThrow (PC.getMessageMetaAndDataById consumer streamName messageId)++-- | Given a message ID, retrieve only the metadata (you can get the data later with 'retrieveDataFromMessageMeta')+getMessageMetaById :: Consumer -> StreamName -> MessageId -> IO MessageMeta+getMessageMetaById consumer streamName messageId = maybeThrow (PC.getMessageMetaById consumer streamName messageId)++-- | Given a message ID, retrieve only the data+getMessageDataById :: Consumer -> StreamName -> MessageId -> IO BS.ByteString+getMessageDataById consumer streamName messageId = maybeThrow (PC.getMessageDataById consumer streamName messageId)++-- | Retrieve the last message in the stream, with data and metadata+getLastMessageMetaAndData :: Consumer -> StreamName -> IO (MessageMeta, BS.ByteString)+getLastMessageMetaAndData consumer streamName = maybeThrow (PC.getLastMessageMetaAndData consumer streamName)++-- | Retrieve the last message in the stream, only metadata (you can get the data later with 'retrieveDataFromMessageMeta')+getLastMessageMeta :: Consumer -> StreamName -> IO MessageMeta+getLastMessageMeta consumer streamName = maybeThrow (PC.getLastMessageMeta consumer streamName)++-- | Retrieve the last message in the stream, only data+getLastMessageData :: Consumer -> StreamName -> IO BS.ByteString+getLastMessageData consumer streamName = maybeThrow (PC.getLastMessageData consumer streamName)++-- | Retrieve the last message in a given stream and group, with data and metadata+getLastInGroupMessageMetaAndData :: Consumer -> StreamName -> GroupId -> IO (MessageMeta, BS.ByteString)+getLastInGroupMessageMetaAndData consumer streamName groupId = maybeThrow (PC.getLastInGroupMessageMetaAndData consumer streamName groupId)++-- | Retrieve the last message in a given stream and group, only metadata (you can get the data later with 'retrieveDataFromMessageMeta')+getLastInGroupMessageMeta :: Consumer -> StreamName -> GroupId -> IO MessageMeta+getLastInGroupMessageMeta consumer streamName groupId = maybeThrow (PC.getLastInGroupMessageMeta consumer streamName groupId)++-- | Retrieve the last message in a given stream and group, only data+getLastInGroupMessageData :: Consumer -> StreamName -> GroupId -> IO BS.ByteString+getLastInGroupMessageData consumer streamName groupId = maybeThrow (PC.getLastInGroupMessageData consumer streamName groupId)++-- | Retrieve the next message in the stream and group, with data and metadata+getNextMessageMetaAndData :: Consumer -> StreamName -> GroupId -> IO (MessageMeta, BS.ByteString)+getNextMessageMetaAndData consumer streamName groupId = maybeThrow (PC.getNextMessageMetaAndData consumer streamName groupId)++-- | Retrieve the next message in the stream and group, only metadata (you can get the data later with 'retrieveDataFromMessageMeta')+getNextMessageMeta :: Consumer -> StreamName -> GroupId -> IO MessageMeta+getNextMessageMeta consumer streamName groupId = maybeThrow (PC.getNextMessageMeta consumer streamName groupId)++-- | 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)++-- | Query messages, return handles without data+queryMessages ::+  Consumer ->+  -- | Actual query string, see the docs for syntax+  Text ->+  StreamName ->+  IO [MessageMeta]+queryMessages consumer query streamName = maybeThrow (PC.queryMessages consumer query streamName)++-- | Retrieve actual data for the handle+retrieveDataForMessageMeta :: Consumer -> MessageMeta -> IO BS.ByteString+retrieveDataForMessageMeta consumer meta = maybeThrow (PC.retrieveDataForMessageMeta consumer meta)

@@ -0,0 +1,198 @@+{-# LANGUAGE BlockArguments #-}+{-# LANGUAGE NamedFieldPuns #-}++module Asapo.Either.Common+  ( SourceType (..),+    InstanceId (..),+    PipelineStep (..),+    Beamline (..),+    withPtr,+    Beamtime (..),+    messageIdFromInt,+    MessageId (..),+    stringHandleToText,+    DataSource (..),+    Token (..),+    timespecToUTC,+    SourceCredentials (..),+    withCredentials,+    withText,+    StreamName (..),+    withCStringN,+    withCStringNToText,+    withConstText,+    withConstCString,+    peekConstCStringText,+    peekCStringText,+    stringHandleToTextUnsafe,+    nominalDiffToMillis,+    retrieveStreamInfoFromC,+    StreamInfo (..),+  )+where++-- \|+-- Description : Utility module with common definitions+--+-- You shouldn't need to explicitly import anything from here+import Asapo.Raw.Common (AsapoSourceCredentialsHandle, AsapoStreamInfoHandle, AsapoStringHandle (AsapoStringHandle), ConstCString, asapo_create_source_credentials, asapo_free_source_credentials, asapo_stream_info_get_finished, asapo_stream_info_get_last_id, asapo_stream_info_get_name, asapo_stream_info_get_next_stream, asapo_stream_info_get_timestamp_created, asapo_stream_info_get_timestamp_last_entry, asapo_string_c_str, kProcessed, kRaw)+import Control.Applicative (pure)+import Control.Exception (bracket)+import Control.Monad (Monad ((>>=)), (>=>))+import Data.Bool (Bool, otherwise)+import Data.Eq ((==))+import Data.Function (($), (.))+import Data.Functor ((<$>))+import Data.Int (Int)+import Data.Maybe (Maybe (Just, Nothing), fromJust)+import Data.Ord ((>))+import Data.String (String)+import Data.Text (Text, pack, unpack)+import Data.Time (NominalDiffTime, zonedTimeToUTC)+import Data.Time.Clock (UTCTime, addUTCTime)+import qualified Data.Time.RFC3339 as RFC3339+import Data.Word (Word64)+import Foreign (Ptr, Storable (peek), nullPtr, with)+import Foreign.C (CChar)+import Foreign.C.ConstPtr (ConstPtr (ConstPtr, unConstPtr))+import Foreign.C.String (CString, peekCString, withCString)+import Foreign.Marshal (alloca, mallocArray)+import Foreign.Marshal.Alloc (free)+import System.Clock (TimeSpec, toNanoSecs)+import System.IO (IO)+import Text.Show (Show)+import Prelude (Fractional ((/)), Integral, Num (fromInteger, (*)), RealFrac (round), fromIntegral)++withText :: Text -> (CString -> IO a) -> IO a+withText t = withCString (unpack t)++nominalDiffToMillis :: (Integral a) => NominalDiffTime -> a+nominalDiffToMillis = round . (* 1000)++newtype MessageId = MessageId Word64 deriving (Show)++messageIdFromInt :: (Integral a) => a -> MessageId+messageIdFromInt = MessageId . fromIntegral++newtype StreamName = StreamName Text deriving (Show)++data SourceType = RawSource | ProcessedSource++newtype InstanceId = InstanceId Text++newtype PipelineStep = PipelineStep Text++newtype Beamtime = Beamtime Text++newtype Beamline = Beamline Text++newtype DataSource = DataSource Text++newtype Token = Token Text++data SourceCredentials = SourceCredentials+  { sourceType :: SourceType,+    instanceId :: InstanceId,+    pipelineStep :: PipelineStep,+    beamtime :: Beamtime,+    beamline :: Beamline,+    dataSource :: DataSource,+    token :: Token+  }++withCredentials :: SourceCredentials -> (AsapoSourceCredentialsHandle -> IO a) -> IO a+withCredentials+  ( SourceCredentials+      { sourceType,+        instanceId = InstanceId instanceId',+        pipelineStep = PipelineStep pipelineStep',+        beamtime = Beamtime beamtime',+        beamline = Beamline beamline',+        dataSource = DataSource dataSource',+        token = Token token'+      }+    )+  f = do+    let convertSourceType RawSource = kRaw+        convertSourceType ProcessedSource = kProcessed+        createCredentialsWithText = withText instanceId' \instanceId'' -> withText pipelineStep' \pipelineStep'' -> withText beamtime' \beamtime'' -> withText beamline' \beamline'' -> withText dataSource' \dataSource'' ->+          withText token' $+            asapo_create_source_credentials+              (convertSourceType sourceType)+              instanceId''+              pipelineStep''+              beamtime''+              beamline''+              dataSource''+    bracket createCredentialsWithText asapo_free_source_credentials f++peekCStringText :: CString -> IO Text+peekCStringText = (pack <$>) . peekCString++peekConstCStringText :: ConstPtr CChar -> IO Text+peekConstCStringText = (pack <$>) . peekCString . unConstPtr++withConstCString :: String -> (ConstCString -> IO b) -> IO b+withConstCString s f = withCString s (f . ConstPtr)++withConstText :: Text -> (ConstCString -> IO a) -> IO a+withConstText t = withConstCString (unpack t)++withCStringN :: Int -> (CString -> IO a) -> IO a+withCStringN size = bracket (mallocArray size) free++withCStringNToText :: Int -> (CString -> IO ()) -> IO Text+withCStringNToText size f =+  withCStringN size \ptr -> do+    f ptr+    pack <$> peekCString ptr++data StreamInfo = StreamInfo+  { streamInfoLastId :: MessageId,+    streamInfoName :: StreamName,+    streamInfoFinished :: Bool,+    streamInfoNextStream :: Text,+    streamInfoCreated :: UTCTime,+    streamInfoLastEntry :: UTCTime+  }+  deriving (Show)++-- Thanks to+--+-- https://github.com/imoverclocked/convert-times/blob/7f9b45bea8e62dbf14a156a8229b68e07efec5a1/app/Main.hs+timespecToUTC :: TimeSpec -> UTCTime+timespecToUTC sc_time =+  let scEpochInUTC :: UTCTime+      scEpochInUTC = zonedTimeToUTC $ fromJust $ RFC3339.parseTimeRFC3339 "1970-01-01T00:00:00.00Z"+      sc2diffTime = fromInteger (toNanoSecs sc_time) / 1000000000 :: NominalDiffTime+   in addUTCTime sc2diffTime scEpochInUTC++retrieveStreamInfoFromC :: AsapoStreamInfoHandle -> IO StreamInfo+retrieveStreamInfoFromC infoHandle = do+  lastId <- asapo_stream_info_get_last_id infoHandle+  name <- asapo_stream_info_get_name infoHandle >>= peekConstCStringText+  nextStream <- asapo_stream_info_get_next_stream infoHandle >>= peekConstCStringText+  finished <- asapo_stream_info_get_finished infoHandle+  created <- alloca \timespecPtr -> do+    asapo_stream_info_get_timestamp_created infoHandle timespecPtr+    timespec <- peek timespecPtr+    pure (timespecToUTC timespec)+  lastEntry <- alloca \timespecPtr -> do+    asapo_stream_info_get_timestamp_last_entry infoHandle timespecPtr+    timespec <- peek timespecPtr+    pure (timespecToUTC timespec)+  pure (StreamInfo (MessageId (fromIntegral lastId)) (StreamName name) (finished > 0) nextStream created lastEntry)++stringHandleToText :: AsapoStringHandle -> IO (Maybe Text)+stringHandleToText handle@(AsapoStringHandle handlePtr)+  | handlePtr == nullPtr = pure Nothing+  | otherwise = Just <$> (asapo_string_c_str handle >>= peekConstCStringText)++stringHandleToTextUnsafe :: AsapoStringHandle -> IO Text+stringHandleToTextUnsafe = asapo_string_c_str >=> peekConstCStringText++withPtr :: (Storable a) => a -> (Ptr a -> IO b) -> IO (a, b)+withPtr h f = with h \hPtr -> do+  result <- f hPtr+  first <- peek hPtr+  pure (first, result)

@@ -0,0 +1,620 @@+{-# LANGUAGE BlockArguments #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}++-- |+-- Description : High-level interface for all consumer-related functions+--+-- To implement an ASAP:O consumer, you should only need this interface.+-- It exposes no memory-management functions (like free) or pointers, and+-- is thus safe to use.+module Asapo.Either.Consumer+  ( FilesystemFlag (..),+    ServerName (..),+    IncludeIncompleteFlag (..),+    SourcePath (..),+    Dataset (..),+    MessageMeta (..),+    DeleteFlag (..),+    ErrorOnNotExistFlag (..),+    Consumer,+    NetworkConnectionType (..),+    StreamFilter (..),+    MessageMetaHandle,+    GroupId,+    Error (..),+    ErrorType (..),+    withConsumer,+    retrieveDataForMessageMeta,+    queryMessages,+    resendNacs,+    getNextDataset,+    retrieveDataFromMeta,+    setStreamPersistent,+    getLastMessageMetaAndData,+    getLastMessageMeta,+    getLastMessageData,+    getLastInGroupMessageMetaAndData,+    getLastInGroupMessageMeta,+    getLastInGroupMessageData,+    getNextMessageMetaAndData,+    getNextMessageMeta,+    getNextMessageData,+    getCurrentDatasetCount,+    getBeamtimeMeta,+    getCurrentSize,+    acknowledge,+    negativeAcknowledge,+    getMessageMetaAndDataById,+    getMessageMetaById,+    getMessageDataById,+    getUnacknowledgedMessages,+    withGroupId,+    queryMessagesHandles,+    setTimeout,+    getLastDataset,+    getLastDatasetInGroup,+    resetLastReadMarker,+    setLastReadMarker,+    getCurrentConnectionType,+    getStreamList,+    deleteStream,+    resolveMetadata,+  )+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.FreeHandleHack (p_asapo_free_handle)+import Control.Applicative (Applicative ((<*>)), pure)+import Control.Exception (bracket)+import Control.Monad (Monad ((>>=)), (>=>))+import Data.Bool (Bool, otherwise)+import qualified Data.ByteString as BS+import Data.Either (Either (Left, Right))+import Data.Eq (Eq ((==)))+import Data.Function (($), (.))+import Data.Functor ((<$>))+import Data.Int (Int)+import Data.Maybe (Maybe (Just, Nothing))+import Data.Ord ((>))+import Data.Text (Text)+import Data.Time (NominalDiffTime, UTCTime)+import Data.Traversable (Traversable (traverse))+import Data.Word (Word64)+import Foreign (Storable (peek), alloca)+import Foreign.C (CInt)+import Foreign.C.ConstPtr (ConstPtr (unConstPtr))+import Foreign.ForeignPtr (ForeignPtr, newForeignPtr, withForeignPtr)+import Foreign.Ptr (Ptr)+import System.IO (IO)+import Text.Show (Show)+import Prelude (Enum, Num ((-)), fromIntegral)++-- | Wrapper for a server name (something like "host:port")+newtype ServerName = ServerName Text++-- | Wrapper for a source path (dubious to not use @FilePath@, but let's see)+newtype SourcePath = SourcePath Text++-- | Whether to use the filesystem or do it in-memory+data FilesystemFlag = WithFilesystem | WithoutFilesystem deriving (Eq)++-- | Wrapper around a consumer handle. Create with the @withConsumer@ function(s).+newtype Consumer = Consumer AsapoConsumerHandle++data ErrorType+  = ErrorNoData+  | ErrorEndOfStream+  | ErrorStreamFinished+  | ErrorUnavailableService+  | ErrorInterruptedTransaction+  | ErrorLocalIOError+  | ErrorWrongInput+  | ErrorPartialData+  | ErrorUnsupportedClient+  | ErrorDataNotInCache+  | ErrorUnknownError+  deriving (Show)++convertErrorType :: AsapoConsumerErrorType -> ErrorType+convertErrorType x | x == kNoData = ErrorNoData+convertErrorType x | x == kEndOfStream = ErrorEndOfStream+convertErrorType x | x == kStreamFinished = ErrorStreamFinished+convertErrorType x | x == kUnavailableService = ErrorUnavailableService+convertErrorType x | x == kInterruptedTransaction = ErrorInterruptedTransaction+convertErrorType x | x == kLocalIOError = ErrorLocalIOError+convertErrorType x | x == kWrongInput = ErrorWrongInput+convertErrorType x | x == kPartialData = ErrorPartialData+convertErrorType x | x == kUnsupportedClient = ErrorUnsupportedClient+convertErrorType x | x == kDataNotInCache = ErrorDataNotInCache+convertErrorType _ = ErrorUnknownError++-- | Wrapper around an ASAP:O producer error, with an additional error code+data Error = Error+  { errorMessage :: Text,+    errorType :: ErrorType+  }+  deriving (Show)++checkErrorWithGivenHandle :: AsapoErrorHandle -> b -> IO (Either Error b)+checkErrorWithGivenHandle errorHandle result = do+  isError <- asapo_is_error errorHandle+  if isError > 0+    then do+      let explanationLength = 1024+      explanation <- withCStringNToText explanationLength \explanationPtr ->+        asapo_error_explain+          errorHandle+          explanationPtr+          (fromIntegral explanationLength)+      errorType' <- asapo_error_get_type errorHandle+      pure (Left (Error explanation (convertErrorType errorType')))+    else pure (Right result)++withErrorHandle :: (AsapoErrorHandle -> IO c) -> IO c+withErrorHandle = bracket asapo_new_error_handle asapo_free_error_handle++checkError :: (Ptr AsapoErrorHandle -> IO b) -> IO (Either Error b)+checkError f = do+  withErrorHandle \errorHandle -> do+    (errorHandlePtr, result) <- withPtr errorHandle f+    checkErrorWithGivenHandle errorHandlePtr result++withSuccess :: (Ptr AsapoErrorHandle -> IO t) -> (t -> IO (Either Error b)) -> IO (Either Error b)+withSuccess toCheck onSuccess = do+  result <- checkError toCheck+  case result of+    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 =+  withCredentials creds \creds' ->+    withConstText serverName \serverNameC ->+      withConstText sourcePath \sourcePathC ->+        checkError (asapo_create_consumer serverNameC sourcePathC (if fsFlag == WithFilesystem then 1 else 0) creds')++-- | 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+  where+    freeConsumer :: Either Error AsapoConsumerHandle -> IO ()+    freeConsumer (Right h) = asapo_free_consumer_handle h+    freeConsumer _ = pure ()+    handle :: Either Error AsapoConsumerHandle -> IO a+    handle (Left e) = onError e+    handle (Right v) = onSuccess (Consumer v)++-- | Wrapper around a group ID+newtype GroupId = GroupId AsapoStringHandle++-- | Allocate a group ID and call a callback+withGroupId :: forall a. Consumer -> (Error -> IO a) -> (GroupId -> IO a) -> IO a+withGroupId (Consumer consumerHandle) onError onSuccess = bracket createGroupId destroy handle+  where+    createGroupId = withSuccess (asapo_consumer_generate_new_group_id consumerHandle) (pure . Right . GroupId)+    destroy (Right (GroupId stringHandle)) = asapo_free_string_handle stringHandle+    destroy _ = pure ()+    handle (Right v) = onSuccess v+    handle (Left e) = onError e++-- | Set the global consumer timeout+setTimeout :: Consumer -> NominalDiffTime -> IO ()+setTimeout (Consumer consumerHandle) timeout = asapo_consumer_set_timeout consumerHandle (nominalDiffToMillis timeout)++-- | Reset the last read marker for a specific group+resetLastReadMarker :: Consumer -> GroupId -> StreamName -> IO (Either Error Int)+resetLastReadMarker (Consumer consumer) (GroupId groupId) (StreamName streamName) =+  withConstText streamName \streamNameC -> (fromIntegral <$>) <$> checkError (asapo_consumer_reset_last_read_marker consumer groupId streamNameC)++-- | Set the last read marker for the stream+setLastReadMarker :: Consumer -> GroupId -> StreamName -> MessageId -> IO (Either Error Int)+setLastReadMarker (Consumer consumer) (GroupId groupId) (StreamName streamName) (MessageId value) =+  withConstText streamName \streamNameC -> (fromIntegral <$>) <$> checkError (asapo_consumer_set_last_read_marker consumer groupId value streamNameC)++-- | Acknowledge a specific message+acknowledge :: Consumer -> GroupId -> StreamName -> MessageId -> IO (Either Error Int)+acknowledge (Consumer consumer) (GroupId groupId) (StreamName streamName) (MessageId messageId) =+  withConstText streamName \streamNameC -> (fromIntegral <$>) <$> checkError (asapo_consumer_acknowledge consumer groupId messageId streamNameC)++-- | Negatively acknowledge a specific message+negativeAcknowledge ::+  Consumer ->+  GroupId ->+  StreamName ->+  MessageId ->+  -- | delay+  NominalDiffTime ->+  IO (Either Error Int)+negativeAcknowledge (Consumer consumer) (GroupId groupId) (StreamName streamName) (MessageId messageId) delay =+  withConstText streamName \streamNameC -> (fromIntegral <$>) <$> checkError (asapo_consumer_negative_acknowledge consumer groupId messageId (nominalDiffToMillis delay) streamNameC)++-- | Get a list of all unacknowledged message IDs in a range+getUnacknowledgedMessages :: Consumer -> GroupId -> StreamName -> (MessageId, MessageId) -> IO (Either Error [MessageId])+getUnacknowledgedMessages (Consumer consumer) (GroupId groupId) (StreamName streamName) (MessageId from, MessageId to) = bracket init destroy handle+  where+    init :: IO (Either Error AsapoIdListHandle)+    init = withConstText streamName $ checkError . asapo_consumer_get_unacknowledged_messages consumer groupId from to+    destroy :: Either Error AsapoIdListHandle -> IO ()+    destroy (Left _) = pure ()+    destroy (Right handle') = asapo_free_id_list_handle handle'+    handle :: Either Error AsapoIdListHandle -> IO (Either Error [MessageId])+    handle (Left e) = pure (Left e)+    handle (Right idListHandle) = do+      numberOfIds <- asapo_id_list_get_size idListHandle+      Right <$> repeatGetterWithSizeLimit ((MessageId <$>) . asapo_id_list_get_item idListHandle) numberOfIds++-- | Network connection type+data NetworkConnectionType+  = -- | not sure about this+    ConnectionUndefined+  | -- | TCP+    ConnectionTcp+  | -- | not sure about this+    ConnectionFabric++convertConnectionType :: AsapoNetworkConnectionType -> NetworkConnectionType+convertConnectionType x | x == kUndefined = ConnectionUndefined+convertConnectionType x | x == kAsapoTcp = ConnectionTcp+convertConnectionType _ = ConnectionFabric++-- | Retrieve the current consumer connection type+getCurrentConnectionType :: Consumer -> IO NetworkConnectionType+getCurrentConnectionType (Consumer consumerHandle) = convertConnectionType <$> asapo_consumer_current_connection_type consumerHandle++data StreamFilter = FilterAllStreams | FilterFinishedStreams | FilterUnfinishedStreams++convertStreamFilter :: StreamFilter -> AsapoStreamFilter+convertStreamFilter FilterAllStreams = kAllStreams+convertStreamFilter FilterFinishedStreams = kFinishedStreams+convertStreamFilter FilterUnfinishedStreams = kUnfinishedStreams++-- An often-repeating pattern: getting a length value and then either+-- returning an empty list or having another function to get the value+-- per ID. With an annoying off-by-one error.+repeatGetterWithSizeLimit :: (Eq a1, Num a1, Applicative f, Enum a1) => (a1 -> f a2) -> a1 -> f [a2]+repeatGetterWithSizeLimit f n+  | n == 0 = pure []+  | otherwise = traverse f [0 .. n - 1]++-- | Retrieve the list of streams with metadata+getStreamList ::+  Consumer ->+  -- | Pass @Nothing@ to get all streams+  Maybe StreamName ->+  StreamFilter ->+  IO (Either Error [StreamInfo])+getStreamList (Consumer consumer) streamName filter = bracket init destroy handle+  where+    init =+      let realStreamName = case streamName of+            Nothing -> ""+            Just (StreamName streamName') -> streamName'+       in withConstText realStreamName $ \streamNameC -> checkError (asapo_consumer_get_stream_list consumer streamNameC (convertStreamFilter filter))+    destroy (Left _) = pure ()+    destroy (Right handle') = asapo_free_stream_infos_handle handle'+    handle (Left e) = pure (Left e)+    handle (Right streamInfosHandle) = do+      numberOfStreams <- asapo_stream_infos_get_size streamInfosHandle+      Right+        <$> repeatGetterWithSizeLimit+          (asapo_stream_infos_get_item streamInfosHandle >=> retrieveStreamInfoFromC)+          numberOfStreams++-- | Anti-boolean-blindness for delete or not delete metadata when deleting a stream+data DeleteFlag = DeleteMeta | DontDeleteMeta deriving (Eq)++-- | Anti-boolean-blindness for "error on not existing data"+data ErrorOnNotExistFlag = ErrorOnNotExist | NoErrorOnNotExist deriving (Eq)++-- | Delete a given stream+deleteStream :: Consumer -> StreamName -> DeleteFlag -> ErrorOnNotExistFlag -> IO (Either Error Int)+deleteStream (Consumer consumer) (StreamName streamName) deleteFlag errorOnNotExistFlag =+  withConstText streamName \streamNameC ->+    (fromIntegral <$>)+      <$> checkError+        ( asapo_consumer_delete_stream+            consumer+            streamNameC+            (if deleteFlag == DeleteMeta then 1 else 0)+            (if errorOnNotExistFlag == ErrorOnNotExist then 1 else 0)+        )++-- | Set a stream persistent+setStreamPersistent :: Consumer -> StreamName -> IO (Either Error Int)+setStreamPersistent (Consumer consumer) (StreamName streamName) =+  withConstText streamName \streamNameC ->+    (fromIntegral <$>)+      <$> checkError+        ( asapo_consumer_set_stream_persistent+            consumer+            streamNameC+        )++-- | Get the current size (number of messages) of the stream+getCurrentSize :: Consumer -> StreamName -> IO (Either Error Int)+getCurrentSize (Consumer consumer) (StreamName streamName) =+  withConstText streamName \streamNameC ->+    (fromIntegral <$>)+      <$> checkError+        ( asapo_consumer_get_current_size+            consumer+            streamNameC+        )++-- | Anti-boolean-blindness for "include incomplete data sets in list"+data IncludeIncompleteFlag = IncludeIncomplete | ExcludeIncomplete deriving (Eq)++-- | Get number of datasets in stream+getCurrentDatasetCount :: Consumer -> StreamName -> IncludeIncompleteFlag -> IO (Either Error Int)+getCurrentDatasetCount (Consumer consumer) (StreamName streamName) inludeIncomplete =+  withConstText streamName \streamNameC ->+    (fromIntegral <$>)+      <$> checkError+        ( asapo_consumer_get_current_dataset_count+            consumer+            streamNameC+            (if inludeIncomplete == IncludeIncomplete then 1 else 0)+        )++-- | Get beamtime metadata (which can be not set, in which case @Nothing@ is returned)+getBeamtimeMeta :: Consumer -> IO (Either Error (Maybe Text))+getBeamtimeMeta (Consumer consumer) = checkError (asapo_consumer_get_beamtime_meta consumer) >>= traverse stringHandleToText++-- | Metadata handle, can be passed around as a pure value and be used to retrieve actual data for the metadata as a two-step process, using the @retrieveDataForMessageMeta@ function(s)+newtype MessageMetaHandle = MessageMetaHandle (ForeignPtr ()) deriving (Show)++withMessageMetaHandle :: MessageMetaHandle -> (AsapoMessageMetaHandle -> IO a) -> IO a+withMessageMetaHandle (MessageMetaHandle foreignPtr) f = withForeignPtr foreignPtr (f . AsapoMessageMetaHandle)++newMessageMetaHandle :: AsapoMessageMetaHandle -> IO MessageMetaHandle+newMessageMetaHandle (AsapoMessageMetaHandle inputPtr) = do+  MessageMetaHandle <$> newForeignPtr p_asapo_free_handle inputPtr++newtype MessageDataHandle = MessageDataHandle (ForeignPtr ()) deriving (Show)++withMessageDataHandle :: MessageDataHandle -> (AsapoMessageDataHandle -> IO a) -> IO a+withMessageDataHandle (MessageDataHandle foreignPtr) f = withForeignPtr foreignPtr (f . AsapoMessageDataHandle)++newMessageDataHandle :: AsapoMessageDataHandle -> IO MessageDataHandle+newMessageDataHandle (AsapoMessageDataHandle inputPtr) =+  MessageDataHandle <$> newForeignPtr p_asapo_free_handle inputPtr++wrapMessageMetaHandle :: AsapoMessageMetaHandle -> IO MessageMetaHandle+wrapMessageMetaHandle = newMessageMetaHandle++-- | Metadata for a single message+data MessageMeta = MessageMeta+  { messageMetaName :: Text,+    messageMetaTimestamp :: UTCTime,+    messageMetaSize :: Word64,+    messageMetaId :: MessageId,+    messageMetaSource :: Text,+    messageMetaMetadata :: Text,+    messageMetaBufId :: Word64,+    messageMetaDatasetSubstream :: Word64,+    messageMetaHandle :: MessageMetaHandle+  }+  deriving (Show)++-- | Retrieve actual data for the handle+retrieveDataForMessageMeta :: Consumer -> MessageMeta -> IO (Either Error BS.ByteString)+retrieveDataForMessageMeta consumer meta = retrieveDataFromMeta consumer (messageMetaHandle meta)++-- | Get the actual metadata hiding behind a handle (shouldn't be necessary when using the convenience interfaces)+resolveMetadata :: MessageMetaHandle -> IO MessageMeta+resolveMetadata metaHandle = withMessageMetaHandle metaHandle \meta -> do+  (timestamp, _) <- withPtr 0 (asapo_message_meta_get_timestamp meta)+  MessageMeta+    <$> (asapo_message_meta_get_name meta >>= peekConstCStringText)+    <*> pure (timespecToUTC timestamp)+    <*> asapo_message_meta_get_size meta+    <*> (MessageId <$> asapo_message_meta_get_id meta)+    <*> (asapo_message_meta_get_source meta >>= peekConstCStringText)+    <*> (asapo_message_meta_get_metadata meta >>= peekConstCStringText)+    <*> asapo_message_meta_get_buf_id meta+    <*> asapo_message_meta_get_dataset_substream meta+    <*> pure metaHandle++-- | Metadata for a dataset+data Dataset = Dataset+  { datasetId :: Word64,+    datasetExpectedSize :: Word64,+    datasetItems :: [MessageMetaHandle]+  }++retrieveDatasetFromC :: AsapoDataSetHandle -> IO Dataset+retrieveDatasetFromC handle = do+  numberOfItems <- asapo_dataset_get_size handle+  items <- repeatGetterWithSizeLimit (asapo_dataset_get_item handle >=> wrapMessageMetaHandle) numberOfItems+  Dataset <$> asapo_dataset_get_id handle <*> asapo_dataset_get_expected_size handle <*> pure items++-- | Get the next dataset for a stream+getNextDataset ::+  Consumer ->+  GroupId ->+  -- | Wait until dataset has these number of messages (0 for maximum size)+  Word64 ->+  StreamName ->+  IO (Either Error Dataset)+getNextDataset (Consumer consumer) (GroupId groupId) minSize (StreamName streamName) = withConstText streamName \streamNameC -> do+  checkError (asapo_consumer_get_next_dataset consumer groupId minSize streamNameC) >>= traverse retrieveDatasetFromC++-- | Get the last dataset in the stream+getLastDataset ::+  Consumer ->+  -- | Wait until dataset has these number of messages (0 for maximum size)+  Word64 ->+  StreamName ->+  IO (Either Error Dataset)+getLastDataset (Consumer consumer) minSize (StreamName streamName) = withConstText streamName \streamNameC -> do+  checkError (asapo_consumer_get_last_dataset consumer minSize streamNameC) >>= traverse retrieveDatasetFromC++-- | Get the last data ste in the given group+getLastDatasetInGroup ::+  Consumer ->+  GroupId ->+  -- | Wait until dataset has these number of messages (0 for maximum size)+  Word64 ->+  StreamName ->+  IO (Either Error Dataset)+getLastDatasetInGroup (Consumer consumer) (GroupId groupId) minSize (StreamName streamName) = withConstText streamName \streamNameC -> do+  checkError (asapo_consumer_get_last_dataset_ingroup consumer groupId minSize streamNameC) >>= traverse retrieveDatasetFromC++retrieveDataFromHandle :: MessageDataHandle -> IO BS.ByteString+retrieveDataFromHandle dataHandle = do+  withMessageDataHandle dataHandle \dataHandlePtr -> do+    messageCString <- asapo_message_data_get_as_chars dataHandlePtr+    BS.packCString (unConstPtr messageCString)++-- | Retrieve data for the given metadata handle+retrieveDataFromMeta :: Consumer -> MessageMetaHandle -> IO (Either Error BS.ByteString)+retrieveDataFromMeta (Consumer consumer) metaHandle =+  withMessageMetaHandle metaHandle \metaHandlePtr ->+    alloca \dataHandlePtrPtr ->+      withSuccess (asapo_consumer_retrieve_data consumer metaHandlePtr dataHandlePtrPtr) \_result -> do+        dataHandlePtr <- peek dataHandlePtrPtr+        dataHandle <- newMessageDataHandle dataHandlePtr+        Right <$> retrieveDataFromHandle dataHandle++withMessageHandles ::+  StreamName ->+  (Ptr AsapoMessageMetaHandle -> Ptr AsapoMessageDataHandle -> ConstCString -> Ptr AsapoErrorHandle -> IO CInt) ->+  (MessageMetaHandle -> MessageDataHandle -> IO (Either Error a)) ->+  IO (Either Error a)+withMessageHandles (StreamName streamName) g f = do+  alloca \metaHandlePtrPtr ->+    alloca \dataHandlePtrPtr ->+      withConstText streamName \streamNameC ->+        withSuccess (g metaHandlePtrPtr dataHandlePtrPtr streamNameC) \_result -> do+          metaHandlePtr <- peek metaHandlePtrPtr+          dataHandlePtr <- peek dataHandlePtrPtr+          metaHandle <- newMessageMetaHandle metaHandlePtr+          dataHandle <- newMessageDataHandle dataHandlePtr+          f metaHandle dataHandle++withMessageHandlesById :: Consumer -> StreamName -> MessageId -> (MessageMetaHandle -> MessageDataHandle -> IO (Either Error a)) -> IO (Either Error a)+withMessageHandlesById (Consumer consumer) streamName (MessageId messageId) =+  withMessageHandles+    streamName+    (asapo_consumer_get_by_id consumer messageId)++retrieveMessageMetaAndData :: MessageMetaHandle -> MessageDataHandle -> IO (Either a (MessageMeta, BS.ByteString))+retrieveMessageMetaAndData metaHandle dataHandle = do+  data' <- retrieveDataFromHandle dataHandle+  meta <- resolveMetadata metaHandle+  pure (Right (meta, data'))++retrieveMessageMeta :: MessageMetaHandle -> p -> IO (Either a MessageMeta)+retrieveMessageMeta metaHandle _dataHandle = do+  meta <- resolveMetadata metaHandle+  pure (Right meta)++retrieveMessageData :: p -> MessageDataHandle -> IO (Either a BS.ByteString)+retrieveMessageData _metaHandle dataHandle = do+  data' <- retrieveDataFromHandle dataHandle+  pure (Right data')++-- | Given a message ID, retrieve both metadata and data+getMessageMetaAndDataById :: Consumer -> StreamName -> MessageId -> IO (Either Error (MessageMeta, BS.ByteString))+getMessageMetaAndDataById consumer streamName messageId =+  withMessageHandlesById consumer streamName messageId retrieveMessageMetaAndData++-- | Given a message ID, retrieve only the metadata (you can get the data later with 'retrieveDataFromMessageMeta')+getMessageMetaById :: Consumer -> StreamName -> MessageId -> IO (Either Error MessageMeta)+getMessageMetaById consumer streamName messageId = do+  withMessageHandlesById consumer streamName messageId retrieveMessageMeta++-- | Given a message ID, retrieve only the data+getMessageDataById :: Consumer -> StreamName -> MessageId -> IO (Either Error BS.ByteString)+getMessageDataById consumer streamName messageId = do+  withMessageHandlesById consumer streamName messageId retrieveMessageData++-- | Retrieve the last message in the stream, with data and metadata+getLastMessageMetaAndData :: Consumer -> StreamName -> IO (Either Error (MessageMeta, BS.ByteString))+getLastMessageMetaAndData (Consumer consumer) streamName = withMessageHandles streamName (asapo_consumer_get_last consumer) retrieveMessageMetaAndData++-- | Retrieve the last message in the stream, only metadata (you can get the data later with 'retrieveDataFromMessageMeta')+getLastMessageMeta :: Consumer -> StreamName -> IO (Either Error MessageMeta)+getLastMessageMeta (Consumer consumer) streamName = withMessageHandles streamName (asapo_consumer_get_last consumer) retrieveMessageMeta++-- | Retrieve the last message in the stream, only data+getLastMessageData :: Consumer -> StreamName -> IO (Either Error BS.ByteString)+getLastMessageData (Consumer consumer) streamName = withMessageHandles streamName (asapo_consumer_get_last consumer) retrieveMessageData++-- | Retrieve the last message in a given stream and group, with data and metadata+getLastInGroupMessageMetaAndData :: Consumer -> StreamName -> GroupId -> IO (Either Error (MessageMeta, BS.ByteString))+getLastInGroupMessageMetaAndData (Consumer consumer) streamName (GroupId groupId) = withMessageHandles streamName (asapo_consumer_get_last_ingroup consumer groupId) retrieveMessageMetaAndData++-- | Retrieve the last message in a given stream and group, only metadata (you can get the data later with 'retrieveDataFromMessageMeta')+getLastInGroupMessageMeta :: Consumer -> StreamName -> GroupId -> IO (Either Error MessageMeta)+getLastInGroupMessageMeta (Consumer consumer) streamName (GroupId groupId) = withMessageHandles streamName (asapo_consumer_get_last_ingroup consumer groupId) retrieveMessageMeta++-- | Retrieve the last message in a given stream and group, only data+getLastInGroupMessageData :: Consumer -> StreamName -> GroupId -> IO (Either Error BS.ByteString)+getLastInGroupMessageData (Consumer consumer) streamName (GroupId groupId) = withMessageHandles streamName (asapo_consumer_get_last_ingroup consumer groupId) retrieveMessageData++-- | Retrieve the next message in the stream and group, with data and metadata+getNextMessageMetaAndData :: Consumer -> StreamName -> GroupId -> IO (Either Error (MessageMeta, BS.ByteString))+getNextMessageMetaAndData (Consumer consumer) streamName (GroupId groupId) =+  withMessageHandles+    streamName+    (asapo_consumer_get_next consumer groupId)+    retrieveMessageMetaAndData++-- | Retrieve the next message in the stream and group, only metadata (you can get the data later with 'retrieveDataFromMessageMeta')+getNextMessageMeta :: Consumer -> StreamName -> GroupId -> IO (Either Error MessageMeta)+getNextMessageMeta (Consumer consumer) streamName (GroupId groupId) = withMessageHandles streamName (asapo_consumer_get_next consumer groupId) retrieveMessageMeta++-- | 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++-- | Query messages, return handles without data+queryMessagesHandles ::+  Consumer ->+  -- | Actual query string, see the docs for syntax+  Text ->+  StreamName ->+  IO (Either Error [MessageMetaHandle])+queryMessagesHandles (Consumer consumer) query (StreamName streamName) = withConstText streamName \streamNameC -> withConstText query \queryC ->+  let init = checkError (asapo_consumer_query_messages consumer queryC streamNameC)+      destroy (Left _) = pure ()+      destroy (Right v) = asapo_free_message_metas_handle v+   in bracket init destroy \case+        Left e -> pure (Left e)+        Right metasHandle' -> do+          numberOfMetas <- asapo_message_metas_get_size metasHandle'+          Right <$> repeatGetterWithSizeLimit (asapo_message_metas_get_item metasHandle' >=> wrapMessageMetaHandle) numberOfMetas++-- | Query messages, return handles without data+queryMessages ::+  Consumer ->+  -- | Actual query string, see the docs for syntax+  Text ->+  StreamName ->+  IO (Either Error [MessageMeta])+queryMessages (Consumer consumer) query (StreamName streamName) = withConstText streamName \streamNameC -> withConstText query \queryC ->+  let init = checkError (asapo_consumer_query_messages consumer queryC streamNameC)+      destroy (Left _) = pure ()+      destroy (Right v) = asapo_free_message_metas_handle v+   in bracket init destroy \case+        Left e -> pure (Left e)+        Right metasHandle' -> do+          numberOfMetas <- asapo_message_metas_get_size metasHandle'+          Right <$> repeatGetterWithSizeLimit (asapo_message_metas_get_item metasHandle' >=> wrapMessageMetaHandle >=> resolveMetadata) numberOfMetas++-- | Reset negative acknowledgements+resendNacs ::+  Consumer ->+  -- | resend yes/no+  Bool ->+  -- | delay+  NominalDiffTime ->+  -- | attempts+  Word64 ->+  IO ()+resendNacs (Consumer consumer) resend delay = asapo_consumer_set_resend_nacs consumer (if resend then 1 else 0) (nominalDiffToMillis delay)

@@ -0,0 +1,688 @@+{-# LANGUAGE BlockArguments #-}+{-# LANGUAGE ScopedTypeVariables #-}++-- |+-- Description : High-level interface for all producer-related functions+--+-- To implement an ASAP:O producer, you should only need this interface.+-- It exposes no memory-management functions (like free) or pointers, and+-- is thus safe to use.+module Asapo.Either.Producer+  ( Endpoint (..),+    ProcessingThreads (..),+    RequestHandlerType (..),+    Error (..),+    Metadata (..),+    SourceCredentials,+    DeletionFlags (..),+    Producer,+    LogLevel (..),+    FileName (..),+    DatasetSubstream (..),+    DatasetSize (..),+    VersionInfo (..),+    UpsertMode (..),+    MetadataIngestMode (..),+    AutoIdFlag (..),+    TransferFlag (..),+    StorageFlag (..),+    RequestResponse (..),+    Opcode (..),+    GenericRequestHeader (..),+    getRequestsQueueSize,+    getRequestsQueueVolumeMb,+    setRequestsQueueLimits,+    checkError,+    checkErrorWithGivenHandle,+    withProducer,+    enableLocalLog,+    waitRequestsFinished,+    getVersionInfo,+    getStreamInfo,+    getStreamMeta,+    getBeamtimeMeta,+    deleteStream,+    getLastStream,+    send,+    sendFile,+    sendStreamFinishedFlag,+    sendBeamtimeMetadata,+    sendStreamMetadata,+    setLogLevel,+    enableRemoteLog,+    setCredentials,+  )+where++import Asapo.Either.Common (MessageId (MessageId), SourceCredentials, StreamInfo, StreamName (StreamName), nominalDiffToMillis, peekCStringText, retrieveStreamInfoFromC, stringHandleToText, stringHandleToTextUnsafe, withCStringNToText, withConstText, withCredentials, withPtr, withText)+import Asapo.Raw.Common (AsapoErrorHandle, AsapoStreamInfoHandle, AsapoStringHandle, asapo_error_explain, asapo_free_error_handle, asapo_free_stream_info_handle, asapo_free_string_handle, asapo_is_error, asapo_new_error_handle, asapo_new_string_handle)+import Asapo.Raw.Producer+  ( AsapoGenericRequestHeader (AsapoGenericRequestHeader),+    AsapoLogLevel,+    AsapoMessageHeaderHandle,+    AsapoOpcode,+    AsapoProducerHandle,+    AsapoRequestCallbackPayloadHandle,+    asapoLogLevelDebug,+    asapoLogLevelError,+    asapoLogLevelInfo,+    asapoLogLevelNone,+    asapoLogLevelWarning,+    asapo_create_message_header,+    asapo_create_producer,+    asapo_free_message_header_handle,+    asapo_free_producer_handle,+    asapo_producer_delete_stream,+    asapo_producer_enable_local_log,+    asapo_producer_enable_remote_log,+    asapo_producer_get_beamtime_meta,+    asapo_producer_get_last_stream,+    asapo_producer_get_requests_queue_size,+    asapo_producer_get_requests_queue_volume_mb,+    asapo_producer_get_stream_info,+    asapo_producer_get_stream_meta,+    asapo_producer_get_version_info,+    asapo_producer_send,+    asapo_producer_send_beamtime_metadata,+    asapo_producer_send_file,+    asapo_producer_send_stream_finished_flag,+    asapo_producer_send_stream_metadata,+    asapo_producer_set_credentials,+    asapo_producer_set_log_level,+    asapo_producer_set_requests_queue_limits,+    asapo_producer_wait_requests_finished,+    asapo_request_callback_payload_get_original_header,+    asapo_request_callback_payload_get_response,+    createRequestCallback,+    kFilesystem,+    kInsert,+    kOpcodeAuthorize,+    kOpcodeCount,+    kOpcodeDeleteStream,+    kOpcodeGetBufferData,+    kOpcodeGetMeta,+    kOpcodeLastStream,+    kOpcodeStreamInfo,+    kOpcodeTransferData,+    kOpcodeTransferDatasetData,+    kOpcodeTransferMetaData,+    kOpcodeUnknownOp,+    kReplace,+    kStoreInDatabase,+    kStoreInFilesystem,+    kTcp,+    kTransferData,+    kTransferMetaDataOnly,+    kUpdate,+  )+import Control.Applicative (Applicative (pure))+import Control.Exception (bracket)+import Data.Bits ((.|.))+import Data.Bool (Bool)+import qualified Data.ByteString as BS+import Data.ByteString.Unsafe (unsafeUseAsCString)+import Data.Either (Either (Left, Right))+import Data.Eq (Eq ((==)))+import Data.Foldable (Foldable (elem))+import Data.Functor ((<$>))+import Data.Int (Int)+import Data.Maybe (Maybe (Just, Nothing))+import Data.Ord ((>))+import Data.Text (Text)+import Data.Time (NominalDiffTime)+import Data.Word (Word64)+import Foreign (Storable (peek), alloca, castPtr)+import Foreign.C.ConstPtr (ConstPtr (unConstPtr))+import Foreign.Ptr (Ptr)+import System.IO (IO)+import Text.Show (Show)+import Prelude (fromIntegral)++-- | Wrapper around an ASAP:O producer error. Note that there is only+--  an error "explanation" here, no error code, since the C interface+--  does not expose this.+newtype Error = Error Text deriving (Show)++-- | Wrapper around an ASAP:O producer endpoint (usually something like "host:port")+newtype Endpoint = Endpoint Text++-- | Wrapper around the number of ASAP:O processing threads (simply to make call signatures mor readable)+newtype ProcessingThreads = ProcessingThreads Int++-- | This has no documentation in ASAP:O yet+data RequestHandlerType = TcpHandler | FilesystemHandler++-- | Opaque wrapper around an ASAP:O producer+newtype Producer = Producer AsapoProducerHandle++-- | Internal function to check and return either an error (if it's+-- present behind the given handle) or a "result" of some function+-- that produced the error handle+checkErrorWithGivenHandle :: AsapoErrorHandle -> b -> IO (Either Error b)+checkErrorWithGivenHandle errorHandle result = do+  isError <- asapo_is_error errorHandle+  if isError > 0+    then do+      let explanationLength = 1024+      explanation <- withCStringNToText explanationLength \explanationPtr ->+        asapo_error_explain+          errorHandle+          explanationPtr+          (fromIntegral explanationLength)+      pure (Left (Error explanation))+    else pure (Right result)++withErrorHandle :: (AsapoErrorHandle -> IO c) -> IO c+withErrorHandle = bracket asapo_new_error_handle asapo_free_error_handle++-- | Helper function since most ASAP:O functions receive a pointer to+-- an error handle as the last argument, so error checking becomes+-- "abstractable"+checkError :: (Ptr AsapoErrorHandle -> IO b) -> IO (Either Error b)+checkError f = do+  withErrorHandle \errorHandle -> 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+  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)+              )++-- | Create a producer and do something with it. This is the main entrypoint into the producer+withProducer ::+  forall a.+  Endpoint ->+  ProcessingThreads ->+  RequestHandlerType ->+  SourceCredentials ->+  -- | timeout+  NominalDiffTime ->+  (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+  where+    freeProducer :: Either Error AsapoProducerHandle -> IO ()+    freeProducer (Left _) = pure ()+    freeProducer (Right producerHandle) = asapo_free_producer_handle producerHandle+    handle :: Either Error AsapoProducerHandle -> IO a+    handle (Left e) = onError e+    handle (Right v) = onSuccess (Producer v)++withStringHandle :: (AsapoStringHandle -> IO c) -> IO c+withStringHandle = bracket asapo_new_string_handle asapo_free_string_handle++data VersionInfo = VersionInfo+  { versionClient :: Text,+    versionServer :: Text,+    versionSupported :: Bool+  }+  deriving (Show)++-- | Retrieve producer version info+getVersionInfo :: Producer -> IO (Either Error VersionInfo)+getVersionInfo (Producer producerHandle) =+  withStringHandle \clientInfo -> withStringHandle \serverInfo -> alloca \supportedPtr -> do+    result <- checkError (asapo_producer_get_version_info producerHandle clientInfo serverInfo supportedPtr)+    case result of+      Left e -> pure (Left e)+      -- The return value is a CInt which is unnecessary probably?+      Right _integerReturnCode -> do+        supported <- peek supportedPtr+        clientInfo' <- stringHandleToTextUnsafe clientInfo+        serverInfo' <- stringHandleToTextUnsafe serverInfo+        pure (Right (VersionInfo clientInfo' serverInfo' (supported > 0)))++-- | Retrieve info for a single stream+getStreamInfo ::+  Producer ->+  StreamName ->+  -- | Timeout+  NominalDiffTime ->+  IO (Either Error StreamInfo)+getStreamInfo (Producer producer) (StreamName stream) timeout = bracket init destroy f+  where+    init :: IO (Either Error AsapoStreamInfoHandle)+    init = withConstText stream \streamC -> checkError (asapo_producer_get_stream_info producer streamC (nominalDiffToMillis timeout))+    destroy :: Either Error AsapoStreamInfoHandle -> IO ()+    destroy (Right handle) = asapo_free_stream_info_handle handle+    destroy _ = pure ()+    f :: Either Error AsapoStreamInfoHandle -> IO (Either Error StreamInfo)+    f (Left e) = pure (Left e)+    f (Right streamInfoHandle) = Right <$> retrieveStreamInfoFromC streamInfoHandle++-- | Retrieve info for the latest stream+getLastStream ::+  Producer ->+  -- | Timeout+  NominalDiffTime ->+  IO (Either Error StreamInfo)+getLastStream (Producer producer) timeout = bracket init destroy f+  where+    init :: IO (Either Error AsapoStreamInfoHandle)+    init = checkError (asapo_producer_get_last_stream producer (nominalDiffToMillis timeout))+    destroy :: Either Error AsapoStreamInfoHandle -> IO ()+    destroy (Right handle) = asapo_free_stream_info_handle handle+    destroy _ = pure ()+    f :: Either Error AsapoStreamInfoHandle -> IO (Either Error StreamInfo)+    f (Left e) = pure (Left e)+    f (Right streamInfoHandle) = Right <$> retrieveStreamInfoFromC streamInfoHandle++-- | Retrieve metadata for the given stream (which might be missing, in which case @Nothing@ is returned)+getStreamMeta ::+  Producer ->+  StreamName ->+  -- | timeout+  NominalDiffTime ->+  IO (Either Error (Maybe Text))+getStreamMeta (Producer producer) (StreamName stream) timeout = bracket init destroy f+  where+    init :: IO (Either Error AsapoStringHandle)+    init = withConstText stream \streamC -> checkError (asapo_producer_get_stream_meta producer streamC (nominalDiffToMillis timeout))+    destroy :: Either Error AsapoStringHandle -> IO ()+    destroy (Right handle) = asapo_free_string_handle handle+    destroy _ = pure ()+    f :: Either Error AsapoStringHandle -> IO (Either Error (Maybe Text))+    f (Left e) = pure (Left e)+    f (Right string) = Right <$> stringHandleToText string++-- | Retrieve metadata for the given stream (which might be missing, in which case @Nothing@ is returned)+getBeamtimeMeta ::+  Producer ->+  -- | timeout+  NominalDiffTime ->+  IO (Either Error (Maybe Text))+getBeamtimeMeta (Producer producer) timeout = bracket init destroy f+  where+    init :: IO (Either Error AsapoStringHandle)+    init = checkError (asapo_producer_get_beamtime_meta producer (nominalDiffToMillis timeout))+    destroy :: Either Error AsapoStringHandle -> IO ()+    destroy (Right handle) = asapo_free_string_handle handle+    destroy _ = pure ()+    f :: Either Error AsapoStringHandle -> IO (Either Error (Maybe Text))+    f (Left e) = pure (Left e)+    f (Right string) = Right <$> stringHandleToText string++data DeletionFlags+  = -- | Delete metadata also+    DeleteMeta+  | -- | Don't throw an error if the data doesn't exist anyways+    DeleteErrorOnNotExist+  deriving (Eq)++-- | Delete the given stream+deleteStream ::+  Producer ->+  StreamName ->+  -- | timeout+  NominalDiffTime ->+  [DeletionFlags] ->+  IO (Either Error Int)+deleteStream (Producer producer) (StreamName stream) timeout deletionFlags = do+  result <- withConstText stream \streamC ->+    checkError+      ( asapo_producer_delete_stream+          producer+          streamC+          (nominalDiffToMillis timeout)+          (if DeleteMeta `elem` deletionFlags then 1 else 0)+          (if DeleteErrorOnNotExist `elem` deletionFlags then 1 else 0)+      )+  pure (fromIntegral <$> result)++-- | Wrapper around file name (dubious to use @Text@ here, but fine for now)+newtype FileName = FileName Text++-- | Wrapper around metadata to be produced+newtype Metadata = Metadata Text++-- | Wrapper around the substream to use+newtype DatasetSubstream = DatasetSubstream Int++-- | Wrapper around the dataset size to use+newtype DatasetSize = DatasetSize Int++-- | Anti-boolean-blindness for the "auto id" flag in the message header+data AutoIdFlag = UseAutoId | NoAutoId deriving (Eq)++-- | Which data to transfer+data TransferFlag = DataAndMetadata | MetadataOnly++-- | Where to store the data+data StorageFlag = Filesystem | Database | FilesystemAndDatabase++convertSendFlags :: TransferFlag -> StorageFlag -> Word64+convertSendFlags tf sf = convertTransferFlag tf .|. convertStorageFlag sf+  where+    convertTransferFlag :: TransferFlag -> Word64+    convertTransferFlag DataAndMetadata = fromIntegral kTransferData+    convertTransferFlag MetadataOnly = fromIntegral kTransferMetaDataOnly+    convertStorageFlag :: StorageFlag -> Word64+    convertStorageFlag Filesystem = fromIntegral kStoreInFilesystem+    convertStorageFlag Database = fromIntegral kStoreInDatabase+    convertStorageFlag FilesystemAndDatabase = fromIntegral kStoreInDatabase .|. fromIntegral kStoreInFilesystem++-- | Internal function to create a message header handle, which gets used to send data.+withMessageHeaderHandle ::+  MessageId ->+  FileName ->+  Metadata ->+  DatasetSubstream ->+  DatasetSize ->+  AutoIdFlag ->+  Int ->+  (AsapoMessageHeaderHandle -> IO b) ->+  IO b+withMessageHeaderHandle+  (MessageId messageId)+  (FileName fileName)+  (Metadata metadata)+  (DatasetSubstream datasetSubstream)+  (DatasetSize datasetSize)+  autoIdFlag+  dataSize = bracket init destroy+    where+      init :: IO AsapoMessageHeaderHandle+      init = withConstText fileName \fileNameC -> withConstText metadata \metadataC ->+        asapo_create_message_header+          messageId+          (fromIntegral dataSize)+          fileNameC+          metadataC+          (fromIntegral datasetSubstream)+          (fromIntegral datasetSize)+          (if autoIdFlag == UseAutoId then 1 else 0)+      destroy = asapo_free_message_header_handle++data Opcode+  = OpcodeUnknownOp+  | OpcodeTransferData+  | OpcodeTransferDatasetData+  | OpcodeStreamInfo+  | OpcodeLastStream+  | OpcodeGetBufferData+  | OpcodeAuthorize+  | OpcodeTransferMetaData+  | OpcodeDeleteStream+  | OpcodeGetMeta+  | OpcodeCount+  | OpcodePersistStream++convertOpcode :: AsapoOpcode -> Opcode+convertOpcode x | x == kOpcodeUnknownOp = OpcodeUnknownOp+convertOpcode x | x == kOpcodeTransferData = OpcodeTransferData+convertOpcode x | x == kOpcodeTransferDatasetData = OpcodeTransferDatasetData+convertOpcode x | x == kOpcodeStreamInfo = OpcodeStreamInfo+convertOpcode x | x == kOpcodeLastStream = OpcodeLastStream+convertOpcode x | x == kOpcodeGetBufferData = OpcodeGetBufferData+convertOpcode x | x == kOpcodeAuthorize = OpcodeAuthorize+convertOpcode x | x == kOpcodeTransferMetaData = OpcodeTransferMetaData+convertOpcode x | x == kOpcodeDeleteStream = OpcodeDeleteStream+convertOpcode x | x == kOpcodeGetMeta = OpcodeGetMeta+convertOpcode x | x == kOpcodeCount = OpcodeCount+convertOpcode _ = OpcodePersistStream++-- | Information about the send request, to be used in the ASAP:O send callback+data GenericRequestHeader = GenericRequestHeader+  { genericRequestHeaderOpCode :: Opcode,+    genericRequestHeaderDataId :: Int,+    genericRequestHeaderDataSize :: Int,+    genericRequestHeaderMetaSize :: Int,+    genericRequestHeaderCustomData :: [Int],+    genericRequestHeaderMessage :: BS.ByteString,+    genericRequestHeaderStream :: Text,+    genericRequestHeaderApiVersion :: Text+  }++convertRequestHeader :: AsapoGenericRequestHeader -> IO GenericRequestHeader+convertRequestHeader (AsapoGenericRequestHeader opcode dataId dataSize metaSize customData message stream apiVersion) = do+  streamText <- peekCStringText stream+  apiVersionText <- peekCStringText apiVersion+  messageAsBs <- BS.packCStringLen (message, fromIntegral dataSize)+  let customData' :: [Int]+      customData' = fromIntegral <$> customData+  pure+    ( GenericRequestHeader+        (convertOpcode opcode)+        (fromIntegral dataId)+        (fromIntegral dataSize)+        (fromIntegral metaSize)+        customData'+        messageAsBs+        streamText+        apiVersionText+    )++-- | Information about the request and its response, to be used in the ASAP:O send callback+data RequestResponse = RequestResponse+  { responsePayload :: Text,+    responseOriginalRequestHeader :: GenericRequestHeader,+    responseError :: Maybe Error+  }++sendRequestCallback :: (RequestResponse -> IO ()) -> Ptr () -> AsapoRequestCallbackPayloadHandle -> AsapoErrorHandle -> IO ()+sendRequestCallback simpleCallback _data 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 ()+  case errorHandle' of+    Left e -> simpleCallback (RequestResponse payloadText originalHeader (Just e))+    _ -> simpleCallback (RequestResponse payloadText originalHeader Nothing)++-- | Send a message containing raw data. Due to newtype and enum usage, all parameter should be self-explanatory+send ::+  Producer ->+  MessageId ->+  FileName ->+  Metadata ->+  DatasetSubstream ->+  DatasetSize ->+  AutoIdFlag ->+  BS.ByteString ->+  TransferFlag ->+  StorageFlag ->+  StreamName ->+  (RequestResponse -> IO ()) ->+  IO (Either Error Int)+send (Producer producer) messageId fileName metadata datasetSubstream datasetSize autoIdFlag data' transferFlag storageFlag (StreamName stream) callback =+  withMessageHeaderHandle+    messageId+    fileName+    metadata+    datasetSubstream+    datasetSize+    autoIdFlag+    (BS.length data')+    \messageHeaderHandle ->+      unsafeUseAsCString data' \data'' -> withConstText stream \streamC -> do+        requestCallback <- createRequestCallback (sendRequestCallback callback)+        ( fromIntegral+            <$>+          )+          <$> checkError+            ( asapo_producer_send+                producer+                messageHeaderHandle+                (castPtr data'')+                (convertSendFlags transferFlag storageFlag)+                streamC+                requestCallback+            )++-- | Send a message containing a file. Due to newtype and enum usage, all parameter should be self-explanatory+sendFile ::+  Producer ->+  MessageId ->+  -- | File name to put into the message header+  FileName ->+  Metadata ->+  DatasetSubstream ->+  DatasetSize ->+  AutoIdFlag ->+  -- | Size+  Int ->+  -- | File to actually send+  FileName ->+  TransferFlag ->+  StorageFlag ->+  StreamName ->+  (RequestResponse -> IO ()) ->+  IO (Either Error Int)+sendFile (Producer producer) messageId fileName meta datasetSubstream datasetSize autoIdFlag size (FileName fileNameToSend) transferFlag storageFlag (StreamName stream) callback =+  withMessageHeaderHandle+    messageId+    fileName+    meta+    datasetSubstream+    datasetSize+    autoIdFlag+    size+    \messageHeaderHandle ->+      withConstText fileNameToSend \fileNameToSendC -> withConstText stream \streamC -> do+        requestCallback <- createRequestCallback (sendRequestCallback callback)+        ( fromIntegral+            <$>+          )+          <$> checkError+            ( asapo_producer_send_file+                producer+                messageHeaderHandle+                fileNameToSendC+                (convertSendFlags transferFlag storageFlag)+                streamC+                requestCallback+            )++-- | As the title says, send the "stream finished" flag+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)+  withConstText stream \streamC -> withConstText nextStream \nextStreamC ->+    (fromIntegral <$>)+      <$> checkError+        ( asapo_producer_send_stream_finished_flag+            producer+            streamC+            lastId+            nextStreamC+            requestCallback+        )++data MetadataIngestMode = Insert | Replace | Update++data UpsertMode = UseUpsert | NoUpsert++-- | Send or extend beamtime metadata+sendBeamtimeMetadata :: Producer -> Metadata -> MetadataIngestMode -> UpsertMode -> (RequestResponse -> IO ()) -> IO (Either Error Int)+sendBeamtimeMetadata (Producer producer) (Metadata metadata) ingestMode upsertMode callback = do+  requestCallback <- createRequestCallback (sendRequestCallback callback)+  (fromIntegral <$>)+    <$> withConstText metadata \metadataC ->+      checkError+        ( asapo_producer_send_beamtime_metadata+            producer+            metadataC+            ( case ingestMode of+                Insert -> kInsert+                Replace -> kReplace+                Update -> kUpdate+            )+            ( case upsertMode of+                UseUpsert -> 1+                _ -> 0+            )+            requestCallback+        )++-- | Send or extend stream metadata+sendStreamMetadata :: Producer -> Metadata -> MetadataIngestMode -> UpsertMode -> StreamName -> (RequestResponse -> IO ()) -> IO (Either Error Int)+sendStreamMetadata (Producer producer) (Metadata metadata) ingestMode upsertMode (StreamName stream) callback = do+  requestCallback <- createRequestCallback (sendRequestCallback callback)+  (fromIntegral <$>)+    <$> withConstText metadata \metadataC -> withConstText stream \streamC ->+      checkError+        ( asapo_producer_send_stream_metadata+            producer+            metadataC+            ( case ingestMode of+                Insert -> kInsert+                Replace -> kReplace+                Update -> kUpdate+            )+            ( case upsertMode of+                UseUpsert -> 1+                _ -> 0+            )+            streamC+            requestCallback+        )++data LogLevel+  = LogNone+  | LogError+  | LogInfo+  | LogDebug+  | LogWarning+  deriving (Eq)++convertLogLevel :: LogLevel -> AsapoLogLevel+convertLogLevel x | x == LogNone = asapoLogLevelNone+convertLogLevel x | x == LogError = asapoLogLevelError+convertLogLevel x | x == LogInfo = asapoLogLevelInfo+convertLogLevel x | x == LogDebug = asapoLogLevelDebug+convertLogLevel _ = asapoLogLevelWarning++-- | Set the log level+setLogLevel :: Producer -> LogLevel -> IO ()+setLogLevel (Producer producer) logLevel = asapo_producer_set_log_level producer (convertLogLevel logLevel)++-- | Enable/Disable logging to stdout+enableLocalLog :: Producer -> Bool -> IO ()+enableLocalLog (Producer producer) enable = asapo_producer_enable_local_log producer (if enable then 1 else 0)++-- | Enable/Disable logging to the central server+enableRemoteLog :: Producer -> Bool -> IO ()+enableRemoteLog (Producer producer) enable = asapo_producer_enable_remote_log producer (if enable then 1 else 0)++-- | Set a different set of credentials+setCredentials :: Producer -> SourceCredentials -> IO (Either Error Int)+setCredentials (Producer producer) credentials = withCredentials credentials \credentialsHandle ->+  (fromIntegral <$>) <$> checkError (asapo_producer_set_credentials producer credentialsHandle)++-- | Get current size of the requests queue (number of requests pending/being processed)+getRequestsQueueSize :: Producer -> IO Int+getRequestsQueueSize (Producer producer) = fromIntegral <$> asapo_producer_get_requests_queue_size producer++-- | Get current volume of the requests queue (total memory of occupied by pending/being processed requests)+getRequestsQueueVolumeMb :: Producer -> IO Int+getRequestsQueueVolumeMb (Producer producer) = fromIntegral <$> asapo_producer_get_requests_queue_volume_mb producer++-- | Set maximum size of the requests queue+setRequestsQueueLimits ::+  Producer ->+  -- | Size (0 for unlimited)+  Int ->+  -- | Volume (in MiB; 0 for unlimited)+  Int ->+  IO ()+setRequestsQueueLimits (Producer producer) size volume =+  asapo_producer_set_requests_queue_limits producer (fromIntegral size) (fromIntegral volume)++-- | Wait for all outstanding requests to finish+waitRequestsFinished :: Producer -> NominalDiffTime -> IO (Either Error Int)+waitRequestsFinished (Producer producer) timeout = do+  (fromIntegral <$>) <$> checkError (asapo_producer_wait_requests_finished producer (nominalDiffToMillis timeout))

@@ -0,0 +1,331 @@+{-# LANGUAGE ScopedTypeVariables #-}+{-# OPTIONS_GHC -Wno-missing-import-lists #-}++-- |+-- Description : High-level interface for all producer-related functions, using exceptions instead of @Either@+--+-- To implement an ASAP:O producer, you should only need this interface.+-- It exposes no memory-management functions (like free) or pointers, and+-- is thus safe to use.+--+-- = Simple Example+--+-- Here's some code for a simple producer that connects and sends a message with id "1".+--+-- >>> :seti -XOverloadedStrings+-- >>> :{+--  import Asapo.Producer+--  import Control.Applicative (Applicative ((<*>)))+--  import Control.Monad(void)+--  import Data.Either (Either (Left, Right))+--  import Data.Function (($))+--  import Data.Functor ((<$>))+--  import Data.Semigroup (Semigroup ((<>)))+--  import Data.Text (Text, pack)+--  import Data.Text.Encoding (encodeUtf8)+--  import qualified Data.Text.IO as TIO+--  import Data.Time.Clock (secondsToNominalDiffTime)+--  import Data.Word (Word64)+--  import System.IO (IO)+--  import Text.Show (Show (show))+--  import Prelude ()+--  main :: IO ()+--  main =+--    withProducer+--      (Endpoint "localhost:8400")+--      (ProcessingThreads 1)+--      TcpHandler+--      ( SourceCredentials+--          { sourceType = RawSource,+--            instanceId = InstanceId "test_instance",+--            pipelineStep = PipelineStep "pipeline_step_1",+--            beamtime = Beamtime "asapo_test",+--            beamline = Beamline "auto",+--            dataSource = DataSource "asapo_source",+--            token = Token "sometoken"+--          }+--      )+--      (secondsToNominalDiffTime 10) $ \producer -> do+--        TIO.putStrLn "connected, sending data"+--        let responseHandler :: RequestResponse -> IO ()+--            responseHandler requestResponse =+--              TIO.putStrLn $ "in response handler, payload "+--                <> responsePayload requestResponse+--                <> ", error "+--                <> pack (show (responseError requestResponse))+--        send+--          producer+--          (MessageId 1)+--          (FileName "raw/default/1.txt")+--          (Metadata "{\"test\": 3.0}")+--          (DatasetSubstream 0)+--          (DatasetSize 0)+--          NoAutoId+--          (encodeUtf8 "test")+--          DataAndMetadata+--          FilesystemAndDatabase+--          (StreamName "default")+--          responseHandler+--        void $ waitRequestsFinished producer (secondsToNominalDiffTime 10)+-- :}+module Asapo.Producer+  ( -- * Types+    ProducerException (..),+    Endpoint (..),+    ProcessingThreads (..),+    RequestHandlerType (..),+    Error (..),+    Metadata (..),+    SourceCredentials (..),+    MessageId (..),+    DeletionFlags (..),+    Producer,+    LogLevel (..),+    FileName (..),+    PipelineStep (..),+    Beamtime (Beamtime),+    DataSource (DataSource),+    Beamline (Beamline),+    StreamInfo (..),+    SourceType (..),+    StreamName (..),+    InstanceId (..),+    Token (..),+    DatasetSubstream (..),+    DatasetSize (..),+    VersionInfo (..),+    UpsertMode (..),+    MetadataIngestMode (..),+    AutoIdFlag (..),+    TransferFlag (..),+    StorageFlag (..),+    RequestResponse (..),+    Opcode (..),+    GenericRequestHeader (..),++    -- * Initialization+    withProducer,++    -- * Getters+    getVersionInfo,+    getStreamInfo,+    getLastStream,+    getBeamtimeMeta,+    getStreamMeta,+    getRequestsQueueSize,+    getRequestsQueueVolumeMb,++    -- * Modifiers+    send,+    sendFile,+    sendStreamFinishedFlag,+    sendBeamtimeMetadata,+    sendStreamMetadata,+    deleteStream,+    setLogLevel,+    enableLocalLog,+    enableRemoteLog,+    setCredentials,+    setRequestsQueueLimits,+    messageIdFromInt,+    waitRequestsFinished,+  )+where++import Asapo.Either.Common+  ( Beamline (Beamline),+    Beamtime (Beamtime),+    DataSource (DataSource),+    InstanceId (..),+    MessageId (..),+    PipelineStep (..),+    SourceCredentials (..),+    SourceType (..),+    StreamInfo (..),+    StreamName (..),+    Token (..),+    messageIdFromInt,+  )+import Asapo.Either.Producer+  ( AutoIdFlag,+    DatasetSize,+    DatasetSubstream,+    DeletionFlags,+    Endpoint,+    Error (Error),+    FileName,+    GenericRequestHeader (..),+    LogLevel (..),+    Metadata,+    MetadataIngestMode,+    Opcode (..),+    ProcessingThreads,+    Producer,+    RequestHandlerType,+    RequestResponse,+    StorageFlag,+    TransferFlag,+    UpsertMode,+    VersionInfo,+    enableLocalLog,+    enableRemoteLog,+    getRequestsQueueSize,+    getRequestsQueueVolumeMb,+    setLogLevel,+    setRequestsQueueLimits,+  )+import qualified Asapo.Either.Producer as PlainProducer+import Control.Applicative (pure)+import Control.Exception (Exception, throw)+import Control.Monad (Monad)+import qualified Data.ByteString as BS+import Data.Either (Either (Left, Right))+import Data.Int (Int)+import Data.Maybe (Maybe)+import Data.Text (Text)+import Data.Time (NominalDiffTime)+import System.IO (IO)+import Text.Show (Show)+import Prelude ()++newtype ProducerException = ProducerException Text deriving (Show)++instance Exception ProducerException++-- | Create a producer and do something with it. This is the main entrypoint into the producer.+withProducer ::+  forall a.+  Endpoint ->+  ProcessingThreads ->+  RequestHandlerType ->+  SourceCredentials ->+  -- | timeout+  NominalDiffTime ->+  (Producer -> IO a) ->+  IO a+withProducer endpoint processingThreads handlerType sourceCredentials timeout onSuccess =+  let onError (Error message) = throw (ProducerException message)+   in PlainProducer.withProducer endpoint processingThreads handlerType sourceCredentials timeout onError onSuccess++maybeThrow :: (Monad m) => m (Either Error b) -> m b+maybeThrow f = do+  result <- f+  case result of+    Left (Error message) -> throw (ProducerException message)+    Right v -> pure v++-- | Retrieve producer version info+getVersionInfo :: Producer -> IO VersionInfo+getVersionInfo producer = maybeThrow (PlainProducer.getVersionInfo producer)++-- | Retrieve info for a single stream+getStreamInfo ::+  Producer ->+  StreamName ->+  -- | Timeout+  NominalDiffTime ->+  IO StreamInfo+getStreamInfo producer stream timeout = maybeThrow (PlainProducer.getStreamInfo producer stream timeout)++-- | Retrieve info for the latest stream+getLastStream ::+  Producer ->+  -- | Timeout+  NominalDiffTime ->+  IO StreamInfo+getLastStream producer timeout = maybeThrow (PlainProducer.getLastStream producer timeout)++-- | Retrieve metadata for the given stream (which might be missing, in which case @Nothing@ is returned)+getStreamMeta ::+  Producer ->+  StreamName ->+  -- | timeout+  NominalDiffTime ->+  IO (Maybe Text)+getStreamMeta producer stream timeout = maybeThrow (PlainProducer.getStreamMeta producer stream timeout)++-- | Retrieve metadata for the given stream (which might be missing, in which case @Nothing@ is returned)+getBeamtimeMeta ::+  Producer ->+  -- | timeout+  NominalDiffTime ->+  IO (Maybe Text)+getBeamtimeMeta producer timeout = maybeThrow (PlainProducer.getBeamtimeMeta producer timeout)++deleteStream ::+  Producer ->+  StreamName ->+  -- | timeout+  NominalDiffTime ->+  [DeletionFlags] ->+  IO Int+deleteStream producer stream timeout deletionFlags = maybeThrow (PlainProducer.deleteStream producer stream timeout deletionFlags)++-- | Send a message containing raw data. Due to newtype and enum usage, all parameter should be self-explanatory+send ::+  Producer ->+  MessageId ->+  FileName ->+  Metadata ->+  DatasetSubstream ->+  DatasetSize ->+  AutoIdFlag ->+  -- | Actual data to send+  BS.ByteString ->+  TransferFlag ->+  StorageFlag ->+  StreamName ->+  (RequestResponse -> IO ()) ->+  IO Int+send producer messageId fileName metadata datasetSubstream datasetSize autoIdFlag data' transferFlag storageFlag stream callback =+  maybeThrow (PlainProducer.send producer messageId fileName metadata datasetSubstream datasetSize autoIdFlag data' transferFlag storageFlag stream callback)++-- | Send a message containing a file. Due to newtype and enum usage, all parameter should be self-explanatory+sendFile ::+  Producer ->+  MessageId ->+  -- | File name to put into the message header+  FileName ->+  Metadata ->+  DatasetSubstream ->+  DatasetSize ->+  AutoIdFlag ->+  -- | Size+  Int ->+  -- | File to actually send+  FileName ->+  TransferFlag ->+  StorageFlag ->+  StreamName ->+  (RequestResponse -> IO ()) ->+  IO Int+sendFile producer messageId fileName meta datasetSubstream datasetSize autoIdFlag size fileNameToSend transferFlag storageFlag stream callback =+  maybeThrow (PlainProducer.sendFile producer messageId fileName meta datasetSubstream datasetSize autoIdFlag size fileNameToSend transferFlag storageFlag stream callback)++-- | As the title says, send the "stream finished" flag+sendStreamFinishedFlag :: Producer -> StreamName -> MessageId -> StreamName -> (RequestResponse -> IO ()) -> IO Int+sendStreamFinishedFlag producer stream lastId nextStream callback =+  maybeThrow+    ( PlainProducer.sendStreamFinishedFlag+        producer+        stream+        lastId+        nextStream+        callback+    )++-- | Send or extend beamtime metadata+sendBeamtimeMetadata :: Producer -> Metadata -> MetadataIngestMode -> UpsertMode -> (RequestResponse -> IO ()) -> IO Int+sendBeamtimeMetadata producer metadata ingestMode upsertMode callback = maybeThrow (PlainProducer.sendBeamtimeMetadata producer metadata ingestMode upsertMode callback)++-- | Send or extend stream metadata+sendStreamMetadata :: Producer -> Metadata -> MetadataIngestMode -> UpsertMode -> StreamName -> (RequestResponse -> IO ()) -> IO Int+sendStreamMetadata producer metadata ingestMode upsertMode stream callback = maybeThrow (PlainProducer.sendStreamMetadata producer metadata ingestMode upsertMode stream callback)++-- | Set a different set of credentials+setCredentials :: Producer -> SourceCredentials -> IO Int+setCredentials producer creds = maybeThrow (PlainProducer.setCredentials producer creds)++-- | Wait for all outstanding requests to finish+waitRequestsFinished :: Producer -> NominalDiffTime -> IO Int+waitRequestsFinished producer timeout = maybeThrow (PlainProducer.waitRequestsFinished producer timeout)

@@ -0,0 +1,169 @@+{-# LANGUAGE BlockArguments #-}+{-# LANGUAGE CApiFFI #-}+{-# LANGUAGE ForeignFunctionInterface #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# OPTIONS_GHC -Wno-unrecognised-pragmas #-}++{-# HLINT ignore "Use camelCase" #-}++module Asapo.Raw.Common+  ( AsapoBool,+    AsapoSourceCredentialsHandle (AsapoSourceCredentialsHandle),+    AsapoMessageDataHandle (AsapoMessageDataHandle),+    asapo_is_error,+    AsapoSourceType,+    asapo_string_c_str,+    mkAsapoFreeWrapper,+    asapo_message_data_get_as_chars,+    asapo_free_stream_infos_handle,+    asapo_new_string_handle,+    asapo_free_string_handle,+    asapo_free_stream_info_handle,+    asapo_free_message_data_handle,+    asapo_error_explain,+    asapo_string_size,+    AsapoErrorHandle (AsapoErrorHandle),+    AsapoStreamInfosHandle (AsapoStreamInfosHandle),+    AsapoStringHandle (AsapoStringHandle),+    ConstCString,+    asapo_free_source_credentials,+    asapo_new_error_handle,+    asapo_free_error_handle,+    asapo_string_from_c_str,+    asapo_new_handle,+    asapo_free_handle,+    asapo_new_message_data_handle,+    asapo_create_source_credentials,+    AsapoStreamInfoHandle (AsapoStreamInfoHandle),+    asapo_stream_infos_get_item,+    asapo_stream_info_get_last_id,+    asapo_stream_info_get_name,+    asapo_stream_info_get_finished,+    asapo_stream_info_get_next_stream,+    asapo_stream_info_get_timestamp_created,+    asapo_stream_info_get_timestamp_last_entry,+    kProcessed,+    kRaw,+  )+where++import Data.Functor ((<$>))+import Foreign (FunPtr, with)+import Foreign.C.ConstPtr (ConstPtr (ConstPtr))+import Foreign.C.String (CString)+import Foreign.C.Types (CChar, CInt (CInt), CSize (CSize), CULong (CULong))+import Foreign.Ptr (Ptr)+import Foreign.Storable (Storable)+import System.Clock (TimeSpec)+import System.IO (IO)+import Prelude ()++-- common+type ConstCString = ConstPtr CChar++type AsapoBool = CInt++newtype {-# CTYPE "asapo/common/common_c.h" "AsapoSourceCredentialsHandle" #-} AsapoSourceCredentialsHandle = AsapoSourceCredentialsHandle (Ptr ()) deriving (Storable)++newtype {-# CTYPE "asapo/common/common_c.h" "AsapoErrorHandle" #-} AsapoErrorHandle = AsapoErrorHandle (Ptr ()) deriving (Storable)++newtype {-# CTYPE "asapo/common/common_c.h" "AsapoStringHandle" #-} AsapoStringHandle = AsapoStringHandle (Ptr ()) deriving (Storable)++asapo_new_string_handle :: IO AsapoStringHandle+asapo_new_string_handle = AsapoStringHandle <$> asapo_new_handle++asapo_free_string_handle :: AsapoStringHandle -> IO ()+asapo_free_string_handle (AsapoStringHandle ptr) = with ptr asapo_free_handle++-- -- FIXME: this is a custom function I added+-- -- foreign import capi "asapo/common/common_c.h &asapo_free_handle___" p_asapo_free_handle :: FunPtr (Ptr () -> IO ())++-- foreign import capi "asapo/common/common_c.h &hs_asapo_free_handle_with_ptr___" p_asapo_free_handle :: FunPtr (Ptr () -> IO ())++newtype {-# CTYPE "asapo/common/common_c.h" "AsapoStreamInfoHandle" #-} AsapoStreamInfoHandle = AsapoStreamInfoHandle (Ptr ()) deriving (Storable)++asapo_free_stream_info_handle :: AsapoStreamInfoHandle -> IO ()+asapo_free_stream_info_handle (AsapoStreamInfoHandle ptr) = with ptr asapo_free_handle++asapo_free_stream_infos_handle :: AsapoStreamInfosHandle -> IO ()+asapo_free_stream_infos_handle (AsapoStreamInfosHandle ptr) = with ptr asapo_free_handle++newtype {-# CTYPE "asapo/common/common_c.h" "AsapoStreamInfosHandle" #-} AsapoStreamInfosHandle = AsapoStreamInfosHandle (Ptr ()) deriving (Storable)++newtype {-# CTYPE "asapo/consumer_c.h" "AsapoMessageDataHandle" #-} AsapoMessageDataHandle = AsapoMessageDataHandle (Ptr ()) deriving (Storable)++asapo_new_message_data_handle :: IO AsapoMessageDataHandle+asapo_new_message_data_handle = AsapoMessageDataHandle <$> asapo_new_handle++asapo_free_message_data_handle :: AsapoMessageDataHandle -> IO ()+asapo_free_message_data_handle (AsapoMessageDataHandle ptr) = with ptr asapo_free_handle++type AsapoSourceType = CInt++foreign import capi "asapo/common/common_c.h value kProcessed" kProcessed :: AsapoSourceType++foreign import capi "asapo/common/common_c.h value kRaw" kRaw :: AsapoSourceType++foreign import capi "asapo/common/common_c.h asapo_free_handle__" asapo_free_handle :: Ptr (Ptr ()) -> IO ()++foreign import ccall "wrapper" mkAsapoFreeWrapper :: (Ptr () -> IO ()) -> IO (FunPtr (Ptr () -> IO ()))++foreign import capi "asapo/common/common_c.h asapo_new_handle" asapo_new_handle :: IO (Ptr ())++asapo_new_error_handle :: IO AsapoErrorHandle+asapo_new_error_handle = AsapoErrorHandle <$> asapo_new_handle++asapo_free_error_handle :: AsapoErrorHandle -> IO ()+asapo_free_error_handle (AsapoErrorHandle ptr) = with ptr asapo_free_handle++foreign import capi "asapo/common/common_c.h asapo_error_explain" asapo_error_explain :: AsapoErrorHandle -> CString -> CSize -> IO ()++foreign import capi "asapo/common/common_c.h asapo_is_error" asapo_is_error :: AsapoErrorHandle -> IO AsapoBool++foreign import capi "asapo/common/common_c.h asapo_string_from_c_str" asapo_string_from_c_str :: ConstCString -> IO AsapoStringHandle++foreign import capi "asapo/common/common_c.h asapo_string_c_str" asapo_string_c_str :: AsapoStringHandle -> IO ConstCString++foreign import capi "asapo/common/common_c.h asapo_string_size" asapo_string_size :: AsapoStringHandle -> IO CSize++foreign import capi "asapo/common/common_c.h asapo_stream_infos_get_item"+  asapo_stream_infos_get_item ::+    AsapoStreamInfosHandle ->+    -- index+    CSize ->+    IO AsapoStreamInfoHandle++foreign import capi "asapo/common/common_c.h asapo_stream_info_get_last_id" asapo_stream_info_get_last_id :: AsapoStreamInfoHandle -> IO CULong++foreign import capi "asapo/common/common_c.h asapo_stream_info_get_name" asapo_stream_info_get_name :: AsapoStreamInfoHandle -> IO ConstCString++-- Yes, this has a typo. Corrected in the Haskell function name+foreign import capi "asapo/common/common_c.h asapo_stream_info_get_ffinished" asapo_stream_info_get_finished :: AsapoStreamInfoHandle -> IO AsapoBool++foreign import capi "asapo/common/common_c.h asapo_stream_info_get_next_stream" asapo_stream_info_get_next_stream :: AsapoStreamInfoHandle -> IO ConstCString++foreign import capi "asapo/common/common_c.h asapo_stream_info_get_timestamp_created" asapo_stream_info_get_timestamp_created :: AsapoStreamInfoHandle -> Ptr TimeSpec -> IO ()++foreign import capi "asapo/common/common_c.h asapo_stream_info_get_timestamp_last_entry" asapo_stream_info_get_timestamp_last_entry :: AsapoStreamInfoHandle -> Ptr TimeSpec -> IO ()++foreign import capi "asapo/common/common_c.h asapo_create_source_credentials"+  asapo_create_source_credentials ::+    AsapoSourceType ->+    -- instance ID+    CString ->+    -- pipeline step+    CString ->+    -- beamtime+    CString ->+    -- beamline+    CString ->+    -- data source+    CString ->+    -- token+    CString ->+    IO AsapoSourceCredentialsHandle++asapo_free_source_credentials :: AsapoSourceCredentialsHandle -> IO ()+asapo_free_source_credentials (AsapoSourceCredentialsHandle ptr) = with ptr asapo_free_handle++foreign import capi "asapo/common/common_c.h asapo_message_data_get_as_chars" asapo_message_data_get_as_chars :: AsapoMessageDataHandle -> IO ConstCString

@@ -0,0 +1,485 @@+{-# LANGUAGE BlockArguments #-}+{-# LANGUAGE CApiFFI #-}+{-# LANGUAGE ForeignFunctionInterface #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# OPTIONS_GHC -Wno-unrecognised-pragmas #-}++{-# HLINT ignore "Use camelCase" #-}++module Asapo.Raw.Consumer+  ( AsapoConsumerHandle (AsapoConsumerHandle),+    AsapoMessageMetaHandle (AsapoMessageMetaHandle),+    AsapoMessageMetasHandle (AsapoMessageMetasHandle),+    AsapoIdListHandle (AsapoIdListHandle),+    AsapoDataSetHandle (AsapoDataSetHandle),+    AsapoPartialErrorDataHandle (AsapoPartialErrorDataHandle),+    AsapoConsumerErrorDataHandle (AsapoConsumerErrorDataHandle),+    AsapoConsumerErrorType,+    asapo_stream_infos_get_size,+    asapo_new_message_meta_handle,+    asapo_free_consumer_handle,+    asapo_free_message_metas_handle,+    asapo_free_id_list_handle,+    kNoData,+    kEndOfStream,+    kStreamFinished,+    kUnavailableService,+    kInterruptedTransaction,+    kLocalIOError,+    kWrongInput,+    kPartialData,+    kUnsupportedClient,+    kDataNotInCache,+    kUnknownError,+    AsapoStreamFilter,+    kAllStreams,+    kFinishedStreams,+    kUnfinishedStreams,+    AsapoNetworkConnectionType,+    kUndefined,+    kAsapoTcp,+    kFabric,+    asapo_error_get_type,+    asapo_create_consumer,+    asapo_consumer_generate_new_group_id,+    asapo_consumer_set_timeout,+    asapo_consumer_reset_last_read_marker,+    asapo_consumer_set_last_read_marker,+    asapo_consumer_acknowledge,+    asapo_consumer_negative_acknowledge,+    asapo_consumer_get_unacknowledged_messages,+    asapo_id_list_get_size,+    asapo_id_list_get_item,+    asapo_consumer_get_last_acknowledged_message,+    asapo_consumer_current_connection_type,+    asapo_consumer_get_stream_list,+    asapo_consumer_delete_stream,+    asapo_consumer_set_stream_persistent,+    asapo_consumer_get_current_size,+    asapo_consumer_get_current_dataset_count,+    asapo_consumer_get_beamtime_meta,+    asapo_consumer_retrieve_data,+    asapo_consumer_get_next_dataset,+    asapo_consumer_get_last_dataset,+    asapo_consumer_get_last_dataset_ingroup,+    asapo_consumer_get_by_id,+    asapo_consumer_get_last,+    asapo_consumer_get_last_ingroup,+    asapo_consumer_get_next,+    asapo_consumer_query_messages,+    asapo_consumer_set_resend_nacs,+    asapo_message_data_get_as_chars,+    asapo_message_meta_get_name,+    asapo_message_meta_get_timestamp,+    asapo_message_meta_get_size,+    asapo_message_meta_get_id,+    asapo_message_meta_get_source,+    asapo_message_meta_get_metadata,+    asapo_message_meta_get_buf_id,+    asapo_message_meta_get_dataset_substream,+    asapo_dataset_get_id,+    asapo_dataset_get_expected_size,+    asapo_dataset_get_size,+    asapo_dataset_get_item,+    asapo_message_metas_get_size,+    asapo_message_metas_get_item,+    asapo_error_get_payload_from_partial_error,+    asapo_partial_error_get_id,+    asapo_partial_error_get_expected_size,+    asapo_error_get_payload_from_consumer_error,+    asapo_consumer_error_get_id,+    asapo_consumer_error_get_next_stream,+  )+where++import Asapo.Raw.Common+  ( AsapoBool,+    AsapoErrorHandle (AsapoErrorHandle),+    AsapoMessageDataHandle (AsapoMessageDataHandle),+    AsapoSourceCredentialsHandle (AsapoSourceCredentialsHandle),+    AsapoStreamInfosHandle (AsapoStreamInfosHandle),+    AsapoStringHandle (AsapoStringHandle),+    ConstCString,+    asapo_free_handle,+    asapo_new_handle,+  )+import Data.Functor ((<$>))+import Data.Int (Int64)+import Data.Word (Word64)+import Foreign.C.ConstPtr (ConstPtr (ConstPtr))+import Foreign.C.Types (CInt (CInt), CLong (CLong), CSize (CSize))+import Foreign.Marshal (with)+import Foreign.Ptr (Ptr)+import Foreign.Storable (Storable)+import System.Clock (TimeSpec)+import System.IO (IO)+import Prelude ()++foreign import capi "asapo/consumer_c.h asapo_stream_infos_get_size" asapo_stream_infos_get_size :: AsapoStreamInfosHandle -> IO CSize++newtype {-# CTYPE "asapo/consumer_c.h" "AsapoConsumerHandle" #-} AsapoConsumerHandle = AsapoConsumerHandle (Ptr ()) deriving (Storable)++newtype {-# CTYPE "asapo/consumer_c.h" "AsapoMessageMetaHandle" #-} AsapoMessageMetaHandle = AsapoMessageMetaHandle (Ptr ()) deriving (Storable)++newtype {-# CTYPE "asapo/consumer_c.h" "AsapoMessageMetasHandle" #-} AsapoMessageMetasHandle = AsapoMessageMetasHandle (Ptr ()) deriving (Storable)++asapo_free_consumer_handle :: AsapoConsumerHandle -> IO ()+asapo_free_consumer_handle (AsapoConsumerHandle ptr) = with ptr asapo_free_handle++asapo_free_message_metas_handle :: AsapoMessageMetasHandle -> IO ()+asapo_free_message_metas_handle (AsapoMessageMetasHandle ptr) = with ptr asapo_free_handle++asapo_new_message_meta_handle :: IO AsapoMessageMetaHandle+asapo_new_message_meta_handle = AsapoMessageMetaHandle <$> asapo_new_handle++newtype {-# CTYPE "asapo/consumer_c.h" "AsapoIdListHandle" #-} AsapoIdListHandle = AsapoIdListHandle (Ptr ()) deriving (Storable)++asapo_free_id_list_handle :: AsapoIdListHandle -> IO ()+asapo_free_id_list_handle (AsapoIdListHandle ptr) = with ptr asapo_free_handle++newtype {-# CTYPE "asapo/consumer_c.h" "AsapoDataSetHandle" #-} AsapoDataSetHandle = AsapoDataSetHandle (Ptr ()) deriving (Storable)++newtype {-# CTYPE "asapo/consumer_c.h" "AsapoPartialErrorDataHandle" #-} AsapoPartialErrorDataHandle = AsapoPartialErrorDataHandle (Ptr ()) deriving (Storable)++newtype {-# CTYPE "asapo/consumer_c.h" "AsapoConsumerErrorDataHandle" #-} AsapoConsumerErrorDataHandle = AsapoConsumerErrorDataHandle (Ptr ()) deriving (Storable)++type AsapoConsumerErrorType = CInt++foreign import capi "asapo/consumer_c.h value kNoData" kNoData :: AsapoConsumerErrorType++foreign import capi "asapo/consumer_c.h value kEndOfStream" kEndOfStream :: AsapoConsumerErrorType++foreign import capi "asapo/consumer_c.h value kStreamFinished" kStreamFinished :: AsapoConsumerErrorType++foreign import capi "asapo/consumer_c.h value kUnavailableService" kUnavailableService :: AsapoConsumerErrorType++foreign import capi "asapo/consumer_c.h value kInterruptedTransaction" kInterruptedTransaction :: AsapoConsumerErrorType++foreign import capi "asapo/consumer_c.h value kLocalIOError" kLocalIOError :: AsapoConsumerErrorType++foreign import capi "asapo/consumer_c.h value kWrongInput" kWrongInput :: AsapoConsumerErrorType++foreign import capi "asapo/consumer_c.h value kPartialData" kPartialData :: AsapoConsumerErrorType++foreign import capi "asapo/consumer_c.h value kUnsupportedClient" kUnsupportedClient :: AsapoConsumerErrorType++foreign import capi "asapo/consumer_c.h value kDataNotInCache" kDataNotInCache :: AsapoConsumerErrorType++foreign import capi "asapo/consumer_c.h value kUnknownError" kUnknownError :: AsapoConsumerErrorType++type AsapoStreamFilter = CInt++foreign import capi "asapo/consumer_c.h value kAllStreams" kAllStreams :: AsapoStreamFilter++foreign import capi "asapo/consumer_c.h value kFinishedStreams" kFinishedStreams :: AsapoStreamFilter++foreign import capi "asapo/consumer_c.h value kUnfinishedStreams" kUnfinishedStreams :: AsapoStreamFilter++type AsapoNetworkConnectionType = CInt++foreign import capi "asapo/consumer_c.h value kUndefined" kUndefined :: AsapoNetworkConnectionType++foreign import capi "asapo/consumer_c.h value kAsapoTcp" kAsapoTcp :: AsapoNetworkConnectionType++foreign import capi "asapo/consumer_c.h value kFabric" kFabric :: AsapoNetworkConnectionType++foreign import capi "asapo/consumer_c.h asapo_error_get_type" asapo_error_get_type :: AsapoErrorHandle -> IO AsapoConsumerErrorType++foreign import capi "asapo/consumer_c.h asapo_create_consumer"+  asapo_create_consumer ::+    -- server_name+    ConstCString ->+    -- source_path+    ConstCString ->+    -- has filesystem+    AsapoBool ->+    AsapoSourceCredentialsHandle ->+    Ptr AsapoErrorHandle ->+    IO AsapoConsumerHandle++foreign import capi "asapo/consumer_c.h asapo_consumer_generate_new_group_id" asapo_consumer_generate_new_group_id :: AsapoConsumerHandle -> Ptr AsapoErrorHandle -> IO AsapoStringHandle++foreign import capi "asapo/consumer_c.h asapo_consumer_set_timeout" asapo_consumer_set_timeout :: AsapoConsumerHandle -> Word64 -> IO ()++foreign import capi "asapo/consumer_c.h asapo_consumer_reset_last_read_marker"+  asapo_consumer_reset_last_read_marker ::+    AsapoConsumerHandle ->+    -- group_id+    AsapoStringHandle ->+    -- stream+    ConstCString ->+    Ptr AsapoErrorHandle ->+    IO CInt++foreign import capi "asapo/consumer_c.h asapo_consumer_set_last_read_marker"+  asapo_consumer_set_last_read_marker ::+    AsapoConsumerHandle ->+    -- group_id+    AsapoStringHandle ->+    -- value+    Word64 ->+    -- stream+    ConstCString ->+    Ptr AsapoErrorHandle ->+    IO CInt++foreign import capi "asapo/consumer_c.h asapo_consumer_acknowledge"+  asapo_consumer_acknowledge ::+    AsapoConsumerHandle ->+    -- group_id+    AsapoStringHandle ->+    -- id+    Word64 ->+    -- stream+    ConstCString ->+    Ptr AsapoErrorHandle ->+    IO CInt++foreign import capi "asapo/consumer_c.h asapo_consumer_negative_acknowledge"+  asapo_consumer_negative_acknowledge ::+    AsapoConsumerHandle ->+    -- group_id+    AsapoStringHandle ->+    -- id+    Word64 ->+    -- delay_ms+    Word64 ->+    -- stream+    ConstCString ->+    Ptr AsapoErrorHandle ->+    IO CInt++foreign import capi "asapo/consumer_c.h asapo_consumer_get_unacknowledged_messages"+  asapo_consumer_get_unacknowledged_messages ::+    AsapoConsumerHandle ->+    -- group_id+    AsapoStringHandle ->+    -- from id+    Word64 ->+    -- to id+    Word64 ->+    -- stream+    ConstCString ->+    Ptr AsapoErrorHandle ->+    IO AsapoIdListHandle++foreign import capi "asapo/consumer_c.h asapo_id_list_get_size" asapo_id_list_get_size :: AsapoIdListHandle -> IO CSize++foreign import capi "asapo/consumer_c.h asapo_id_list_get_item" asapo_id_list_get_item :: AsapoIdListHandle -> CSize -> IO Word64++foreign import capi "asapo/consumer_c.h asapo_consumer_get_last_acknowledged_message"+  asapo_consumer_get_last_acknowledged_message ::+    AsapoConsumerHandle ->+    -- group_id+    AsapoStringHandle ->+    -- stream+    ConstCString ->+    Ptr AsapoErrorHandle ->+    IO CLong++-- foreign import capi "asapo/consumer_c.h asapo_consumer_force_no_rdma" asapo_consumer_force_no_rdma :: AsapoConsumerHandle -> IO ()++foreign import capi "asapo/consumer_c.h asapo_consumer_current_connection_type" asapo_consumer_current_connection_type :: AsapoConsumerHandle -> IO AsapoNetworkConnectionType++-- Again, a typo in the original (no "asapo_" prefix, but corrected in the Haskell code)+-- foreign import capi "asapo/consumer_c.h enable_new_monitoring_api_format" asapo_enable_new_monitoring_api_format :: AsapoConsumerHandle -> AsapoBool -> Ptr AsapoErrorHandle -> IO CInt++foreign import capi "asapo/consumer_c.h asapo_consumer_get_stream_list"+  asapo_consumer_get_stream_list ::+    AsapoConsumerHandle ->+    -- stream+    ConstCString ->+    AsapoStreamFilter ->+    Ptr AsapoErrorHandle ->+    IO AsapoStreamInfosHandle++foreign import capi "asapo/consumer_c.h asapo_consumer_delete_stream"+  asapo_consumer_delete_stream ::+    AsapoConsumerHandle ->+    -- stream+    ConstCString ->+    -- delete_meta+    AsapoBool ->+    -- error_on_not_exist+    AsapoBool ->+    Ptr AsapoErrorHandle ->+    IO CInt++foreign import capi "asapo/consumer_c.h asapo_consumer_set_stream_persistent"+  asapo_consumer_set_stream_persistent ::+    AsapoConsumerHandle ->+    -- stream+    ConstCString ->+    Ptr AsapoErrorHandle ->+    IO CInt++foreign import capi "asapo/consumer_c.h asapo_consumer_get_current_size"+  asapo_consumer_get_current_size ::+    AsapoConsumerHandle ->+    -- stream+    ConstCString ->+    Ptr AsapoErrorHandle ->+    IO Int64++foreign import capi "asapo/consumer_c.h asapo_consumer_get_current_dataset_count"+  asapo_consumer_get_current_dataset_count ::+    AsapoConsumerHandle ->+    -- stream+    ConstCString ->+    -- include_incomplete+    AsapoBool ->+    Ptr AsapoErrorHandle ->+    IO Int64++foreign import capi "asapo/consumer_c.h asapo_consumer_get_beamtime_meta" asapo_consumer_get_beamtime_meta :: AsapoConsumerHandle -> Ptr AsapoErrorHandle -> IO AsapoStringHandle++foreign import capi "asapo/consumer_c.h asapo_consumer_retrieve_data"+  asapo_consumer_retrieve_data ::+    AsapoConsumerHandle ->+    AsapoMessageMetaHandle ->+    Ptr AsapoMessageDataHandle ->+    Ptr AsapoErrorHandle ->+    IO CInt++foreign import capi "asapo/consumer_c.h asapo_consumer_get_next_dataset"+  asapo_consumer_get_next_dataset ::+    AsapoConsumerHandle ->+    -- group_id+    AsapoStringHandle ->+    -- min_size+    Word64 ->+    -- stream+    ConstCString ->+    Ptr AsapoErrorHandle ->+    IO AsapoDataSetHandle++foreign import capi "asapo/consumer_c.h asapo_consumer_get_last_dataset"+  asapo_consumer_get_last_dataset ::+    AsapoConsumerHandle ->+    -- min_size+    Word64 ->+    -- stream+    ConstCString ->+    Ptr AsapoErrorHandle ->+    IO AsapoDataSetHandle++foreign import capi "asapo/consumer_c.h asapo_consumer_get_last_dataset_ingroup"+  asapo_consumer_get_last_dataset_ingroup ::+    AsapoConsumerHandle ->+    -- group_id+    AsapoStringHandle ->+    -- min_size+    Word64 ->+    -- stream+    ConstCString ->+    Ptr AsapoErrorHandle ->+    IO AsapoDataSetHandle++foreign import capi "asapo/consumer_c.h asapo_consumer_get_by_id"+  asapo_consumer_get_by_id ::+    AsapoConsumerHandle ->+    -- id+    Word64 ->+    Ptr AsapoMessageMetaHandle ->+    Ptr AsapoMessageDataHandle ->+    -- stream+    ConstCString ->+    Ptr AsapoErrorHandle ->+    IO CInt++foreign import capi "asapo/consumer_c.h asapo_consumer_get_last"+  asapo_consumer_get_last ::+    AsapoConsumerHandle ->+    Ptr AsapoMessageMetaHandle ->+    Ptr AsapoMessageDataHandle ->+    -- stream+    ConstCString ->+    Ptr AsapoErrorHandle ->+    IO CInt++foreign import capi "asapo/consumer_c.h asapo_consumer_get_last_ingroup"+  asapo_consumer_get_last_ingroup ::+    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"+  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_query_messages"+  asapo_consumer_query_messages ::+    AsapoConsumerHandle ->+    -- query+    ConstCString ->+    -- stream+    ConstCString ->+    Ptr AsapoErrorHandle ->+    IO AsapoMessageMetasHandle++foreign import capi "asapo/consumer_c.h asapo_consumer_set_resend_nacs"+  asapo_consumer_set_resend_nacs ::+    AsapoConsumerHandle ->+    -- resend+    AsapoBool ->+    -- delay_ms+    Word64 ->+    -- resend_attempts+    Word64 ->+    IO ()++foreign import capi "asapo/consumer_c.h asapo_message_data_get_as_chars" asapo_message_data_get_as_chars :: AsapoMessageDataHandle -> IO ConstCString++foreign import capi "asapo/consumer_c.h asapo_message_meta_get_name" asapo_message_meta_get_name :: AsapoMessageMetaHandle -> IO ConstCString++foreign import capi "asapo/consumer_c.h asapo_message_meta_get_timestamp" asapo_message_meta_get_timestamp :: AsapoMessageMetaHandle -> Ptr TimeSpec -> IO ()++foreign import capi "asapo/consumer_c.h asapo_message_meta_get_size" asapo_message_meta_get_size :: AsapoMessageMetaHandle -> IO Word64++foreign import capi "asapo/consumer_c.h asapo_message_meta_get_id" asapo_message_meta_get_id :: AsapoMessageMetaHandle -> IO Word64++foreign import capi "asapo/consumer_c.h asapo_message_meta_get_source" asapo_message_meta_get_source :: AsapoMessageMetaHandle -> IO ConstCString++foreign import capi "asapo/consumer_c.h asapo_message_meta_get_metadata" asapo_message_meta_get_metadata :: AsapoMessageMetaHandle -> IO ConstCString++foreign import capi "asapo/consumer_c.h asapo_message_meta_get_buf_id" asapo_message_meta_get_buf_id :: AsapoMessageMetaHandle -> IO Word64++foreign import capi "asapo/consumer_c.h asapo_message_meta_get_dataset_substream" asapo_message_meta_get_dataset_substream :: AsapoMessageMetaHandle -> IO Word64++foreign import capi "asapo/consumer_c.h asapo_dataset_get_id" asapo_dataset_get_id :: AsapoDataSetHandle -> IO Word64++foreign import capi "asapo/consumer_c.h asapo_dataset_get_expected_size" asapo_dataset_get_expected_size :: AsapoDataSetHandle -> IO Word64++foreign import capi "asapo/consumer_c.h asapo_dataset_get_size" asapo_dataset_get_size :: AsapoDataSetHandle -> IO CSize++foreign import capi "asapo/consumer_c.h asapo_dataset_get_item" asapo_dataset_get_item :: AsapoDataSetHandle -> CSize -> IO AsapoMessageMetaHandle++foreign import capi "asapo/consumer_c.h asapo_message_metas_get_size" asapo_message_metas_get_size :: AsapoMessageMetasHandle -> IO CSize++foreign import capi "asapo/consumer_c.h asapo_message_metas_get_item" asapo_message_metas_get_item :: AsapoMessageMetasHandle -> CSize -> IO AsapoMessageMetaHandle++foreign import capi "asapo/consumer_c.h asapo_error_get_payload_from_partial_error" asapo_error_get_payload_from_partial_error :: AsapoErrorHandle -> IO AsapoPartialErrorDataHandle++foreign import capi "asapo/consumer_c.h asapo_partial_error_get_id" asapo_partial_error_get_id :: AsapoPartialErrorDataHandle -> IO Word64++foreign import capi "asapo/consumer_c.h asapo_partial_error_get_expected_size" asapo_partial_error_get_expected_size :: AsapoPartialErrorDataHandle -> IO Word64++foreign import capi "asapo/consumer_c.h asapo_error_get_payload_from_consumer_error" asapo_error_get_payload_from_consumer_error :: AsapoErrorHandle -> IO AsapoConsumerErrorDataHandle++foreign import capi "asapo/consumer_c.h asapo_consumer_error_get_id" asapo_consumer_error_get_id :: AsapoConsumerErrorDataHandle -> IO Word64++foreign import capi "asapo/consumer_c.h asapo_consumer_error_get_next_stream" asapo_consumer_error_get_next_stream :: AsapoConsumerErrorDataHandle -> IO ConstCString

@@ -0,0 +1,19 @@+{-# LANGUAGE BlockArguments #-}+{-# LANGUAGE CApiFFI #-}+{-# LANGUAGE ForeignFunctionInterface #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}++#include <asapo/common/common_c.h>++-- |+-- Description : Here be dragons+--+-- Actually, not really dragons, just a small hack in order to free up memory without patching asapo or writing a whole C module+module Asapo.Raw.FreeHandleHack(p_asapo_free_handle) where+import Foreign(Ptr, FunPtr)+import Prelude()+import System.IO(IO)++#def void hs_asapo_free_handle_with_ptr(void *handle) { asapo_free_handle__(&handle); }++foreign import capi "asapo/common/common_c.h &hs_asapo_free_handle_with_ptr" p_asapo_free_handle :: FunPtr (Ptr () -> IO ())

@@ -0,0 +1,425 @@+{-# LANGUAGE CApiFFI #-}+{-# LANGUAGE BlockArguments #-}+{-# LANGUAGE ForeignFunctionInterface #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}++-- hsc2hs needs the include to find the C structure+#include <asapo/producer_c.h>++module Asapo.Raw.Producer+  ( AsapoProducerHandle (AsapoProducerHandle),+    AsapoRequestCallbackPayloadHandle (AsapoRequestCallbackPayloadHandle),+    AsapoMessageHeaderHandle (AsapoMessageHeaderHandle),+    AsapoRequestCallback,+    createRequestCallback,+    kMaxMessageSize,+    asapo_free_producer_handle,+    kNCustomParams,+    kDefaultIngestMode,+    kMaxVersionSize,+    asapo_free_message_header_handle,+    AsapoRequestHandlerType,+    kTcp,+    kFilesystem,+    AsapoIngestModeFlags,+    kTransferData,+    AsapoOpcode,+    kOpcodeUnknownOp,+    kOpcodeTransferData,+    kOpcodeTransferDatasetData,+    kOpcodeStreamInfo,+    kOpcodeLastStream,+    kOpcodeGetBufferData,+    kOpcodeAuthorize,+    kOpcodeTransferMetaData,+    kOpcodeDeleteStream,+    kOpcodeGetMeta,+    kOpcodeCount,+    kOpcodePersistStream,+    kTransferMetaDataOnly,+    kStoreInFilesystem,+    kStoreInDatabase,+    AsapoMetaIngestOp,+    kInsert,+    kReplace,+    kUpdate,+    AsapoLogLevel,+    asapoLogLevelNone,+    asapoLogLevelError,+    asapoLogLevelInfo,+    asapoLogLevelDebug,+    asapoLogLevelWarning,+    asapo_create_producer,+    asapo_producer_get_version_info,+    asapo_producer_get_stream_info,+    asapo_producer_get_stream_meta,+    asapo_producer_get_beamtime_meta,+    asapo_producer_delete_stream,+    asapo_producer_get_last_stream,+    asapo_create_message_header,+    asapo_producer_send,+    asapo_producer_send_file,+    asapo_producer_send_stream_finished_flag,+    asapo_producer_send_beamtime_metadata,+    asapo_producer_send_stream_metadata,+    asapo_request_callback_payload_get_response,+    asapo_request_callback_payload_get_original_header,+    asapo_producer_set_log_level,+    asapo_producer_enable_local_log,+    asapo_producer_enable_remote_log,+    asapo_producer_set_credentials,+    asapo_producer_get_requests_queue_size,+    asapo_producer_get_requests_queue_volume_mb,+    asapo_producer_set_requests_queue_limits,+    asapo_producer_wait_requests_finished,+    AsapoGenericRequestHeader(..)+  )+where++import Data.Functor((<$>))+import Data.Word(Word64)+import Asapo.Raw.Common+  ( AsapoBool,+    AsapoErrorHandle (AsapoErrorHandle),+    AsapoSourceCredentialsHandle (AsapoSourceCredentialsHandle),+    AsapoStreamInfoHandle (AsapoStreamInfoHandle),+    AsapoStringHandle (AsapoStringHandle),+    ConstCString,+    asapo_free_handle+  )+import Foreign.C.ConstPtr (ConstPtr (ConstPtr))+import Foreign.C.String (CString)+import Foreign.C.Types (CInt (CInt), CSize (CSize), CUChar (CUChar))+import Foreign.Ptr (FunPtr, Ptr, plusPtr)+import Foreign (with, peekArray)+import Foreign.Storable (Storable(alignment, peek, peekByteOff, poke, sizeOf))+import System.IO (IO)+import Prelude (error, fromIntegral)+import Control.Applicative((<*>), pure)++newtype {-# CTYPE "asapo/producer_c.h" "AsapoProducerHandle" #-} AsapoProducerHandle = AsapoProducerHandle (Ptr ()) deriving (Storable)++asapo_free_producer_handle :: AsapoProducerHandle -> IO ()+asapo_free_producer_handle (AsapoProducerHandle ptr) = with ptr \ptr' -> asapo_free_handle ptr'++newtype {-# CTYPE "asapo/producer_c.h" "AsapoRequestCallbackPayloadHandle" #-} AsapoRequestCallbackPayloadHandle = AsapoRequestCallbackPayloadHandle (Ptr ()) deriving (Storable)++newtype {-# CTYPE "asapo/producer_c.h" "AsapoMessageHeaderHandle" #-} AsapoMessageHeaderHandle = AsapoMessageHeaderHandle (Ptr ()) deriving (Storable)++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 ()++foreign import ccall "wrapper" createRequestCallback :: AsapoRequestCallback -> IO (FunPtr AsapoRequestCallback)++foreign import capi "asapo/producer_c.h value kMaxMessageSize" kMaxMessageSize :: CSize++foreign import capi "asapo/producer_c.h value kMaxVersionSize" kMaxVersionSize :: CSize++foreign import capi "asapo/producer_c.h value kNCustomParams" kNCustomParams :: CSize++type AsapoOpcode = CInt++foreign import capi "asapo/producer_c.h value kOpcodeUnknownOp" kOpcodeUnknownOp :: AsapoOpcode++foreign import capi "asapo/producer_c.h value kOpcodeTransferData" kOpcodeTransferData :: AsapoOpcode++foreign import capi "asapo/producer_c.h value kOpcodeTransferDatasetData" kOpcodeTransferDatasetData :: AsapoOpcode++foreign import capi "asapo/producer_c.h value kOpcodeStreamInfo" kOpcodeStreamInfo :: AsapoOpcode++foreign import capi "asapo/producer_c.h value kOpcodeLastStream" kOpcodeLastStream :: AsapoOpcode++foreign import capi "asapo/producer_c.h value kOpcodeGetBufferData" kOpcodeGetBufferData :: AsapoOpcode++foreign import capi "asapo/producer_c.h value kOpcodeAuthorize" kOpcodeAuthorize :: AsapoOpcode++foreign import capi "asapo/producer_c.h value kOpcodeTransferMetaData" kOpcodeTransferMetaData :: AsapoOpcode++foreign import capi "asapo/producer_c.h value kOpcodeDeleteStream" kOpcodeDeleteStream :: AsapoOpcode++foreign import capi "asapo/producer_c.h value kOpcodeGetMeta" kOpcodeGetMeta :: AsapoOpcode++foreign import capi "asapo/producer_c.h value kOpcodeCount" kOpcodeCount :: AsapoOpcode++foreign import capi "asapo/producer_c.h value kOpcodePersistStream" kOpcodePersistStream :: AsapoOpcode++data AsapoGenericRequestHeader = AsapoGenericRequestHeader+  { asapoGenericRequestHeaderOpCode :: !AsapoOpcode,+    asapoGenericRequestHeaderDataId :: !Word64,+    asapoGenericRequestHeaderDataSize :: !Word64,+    asapoGenericRequestHeaderMetaSize :: !Word64,+    asapoGenericRequestHeaderCustomData :: ![Word64],+    asapoGenericRequestHeaderMessage :: !CString,+    asapoGenericRequestHeaderStream :: !CString,+    asapoGenericRequestHeaderApiVersion :: !CString+  }++instance Storable AsapoGenericRequestHeader where+  sizeOf _ = (# size struct AsapoGenericRequestHeader)+  alignment _ = (# alignment struct AsapoGenericRequestHeader)+  peek ptr = do+    customData <- peekArray (fromIntegral kNCustomParams) ((# ptr struct AsapoGenericRequestHeader, custom_data) ptr)+    opCode <- (# peek struct AsapoGenericRequestHeader, op_code) ptr+    dataId <- (# peek struct AsapoGenericRequestHeader, data_id) ptr+    dataSize <- (# peek struct AsapoGenericRequestHeader, data_size) ptr+    metaSize <- (# peek struct AsapoGenericRequestHeader, meta_size) ptr+    let message' = (# ptr struct AsapoGenericRequestHeader, message) ptr+    let stream' = (# ptr struct AsapoGenericRequestHeader, stream) ptr+    let apiVersion' = (# ptr struct AsapoGenericRequestHeader, api_version) ptr+    AsapoGenericRequestHeader+      <$> pure opCode+      <*> pure dataId+      <*> pure dataSize+      <*> pure metaSize+      <*> pure customData+      <*> pure message'+      <*> pure stream'+      <*> pure apiVersion'+  poke _ = error "why was AsapoGenericRequestHeader poked? it's supposed to be read-only"+++type AsapoRequestHandlerType = CInt++foreign import capi "asapo/producer_c.h value kTcp" kTcp :: AsapoRequestHandlerType++foreign import capi "asapo/producer_c.h value kFilesystem" kFilesystem :: AsapoRequestHandlerType++type AsapoIngestModeFlags = CInt++foreign import capi "asapo/producer_c.h value kTransferData" kTransferData :: AsapoIngestModeFlags++foreign import capi "asapo/producer_c.h value kTransferMetaDataOnly" kTransferMetaDataOnly :: AsapoIngestModeFlags++foreign import capi "asapo/producer_c.h value kStoreInFilesystem" kStoreInFilesystem :: AsapoIngestModeFlags++foreign import capi "asapo/producer_c.h value kStoreInDatabase" kStoreInDatabase :: AsapoIngestModeFlags++foreign import capi "asapo/producer_c.h value kDefaultIngestMode" kDefaultIngestMode :: AsapoIngestModeFlags++type AsapoMetaIngestOp = CInt++foreign import capi "asapo/producer_c.h value kInsert" kInsert :: AsapoMetaIngestOp++foreign import capi "asapo/producer_c.h value kReplace" kReplace :: AsapoMetaIngestOp++foreign import capi "asapo/producer_c.h value kUpdate" kUpdate :: AsapoMetaIngestOp++type AsapoLogLevel = CInt++foreign import capi "asapo/producer_c.h value None" asapoLogLevelNone :: AsapoLogLevel++foreign import capi "asapo/producer_c.h value Error" asapoLogLevelError :: AsapoLogLevel++foreign import capi "asapo/producer_c.h value Info" asapoLogLevelInfo :: AsapoLogLevel++foreign import capi "asapo/producer_c.h value Debug" asapoLogLevelDebug :: AsapoLogLevel++foreign import capi "asapo/producer_c.h value Warning" asapoLogLevelWarning :: AsapoLogLevel++foreign import capi "asapo/producer_c.h asapo_create_producer"+  asapo_create_producer ::+    -- endpoint+    CString ->+    -- processing threads+    CUChar ->+    AsapoRequestHandlerType ->+    AsapoSourceCredentialsHandle ->+    -- timeout_ms+    Word64 ->+    Ptr AsapoErrorHandle ->+    IO AsapoProducerHandle++foreign import capi "asapo/producer_c.h asapo_producer_get_version_info"+  asapo_producer_get_version_info ::+    AsapoProducerHandle ->+    -- client info+    AsapoStringHandle ->+    -- server info+    AsapoStringHandle ->+    -- supported+    Ptr AsapoBool ->+    Ptr AsapoErrorHandle ->+    IO CInt++foreign import capi "asapo/producer_c.h asapo_producer_get_stream_info"+  asapo_producer_get_stream_info ::+    AsapoProducerHandle ->+    -- stream+    ConstCString ->+    -- timeout ms+    Word64 ->+    Ptr AsapoErrorHandle ->+    IO AsapoStreamInfoHandle++foreign import capi "asapo/producer_c.h asapo_producer_get_stream_meta"+  asapo_producer_get_stream_meta ::+    AsapoProducerHandle ->+    -- stream+    ConstCString ->+    -- timeout_ms+    Word64 ->+    Ptr AsapoErrorHandle ->+    IO AsapoStringHandle++foreign import capi "asapo/producer_c.h asapo_producer_get_beamtime_meta"+  asapo_producer_get_beamtime_meta ::+    AsapoProducerHandle ->+    -- timeout_ms+    Word64 ->+    Ptr AsapoErrorHandle ->+    IO AsapoStringHandle++foreign import capi "asapo/producer_c.h asapo_producer_delete_stream"+  asapo_producer_delete_stream ::+    AsapoProducerHandle ->+    -- stream+    ConstCString ->+    -- timeout_ms+    Word64 ->+    -- delete meta+    AsapoBool ->+    -- error_on_not_exist+    AsapoBool ->+    Ptr AsapoErrorHandle ->+    IO CInt++foreign import capi "asapo/producer_c.h asapo_producer_get_last_stream"+  asapo_producer_get_last_stream ::+    AsapoProducerHandle ->+    -- timeout ms+    Word64 ->+    Ptr AsapoErrorHandle ->+    IO AsapoStreamInfoHandle++foreign import capi "asapo/producer_c.h asapo_create_message_header"+  asapo_create_message_header ::+    -- message ID+    Word64 ->+    -- data size+    Word64 ->+    -- file name+    ConstCString ->+    -- user metadata+    ConstCString ->+    -- dataset substream+    Word64 ->+    -- dataset size+    Word64 ->+    -- auto id+    AsapoBool ->+    IO AsapoMessageHeaderHandle++foreign import capi "asapo/producer_c.h asapo_producer_send"+  asapo_producer_send ::+    AsapoProducerHandle ->+    AsapoMessageHeaderHandle ->+    -- data+    Ptr () ->+    -- ingest mode+    Word64 ->+    -- stream+    ConstCString ->+    FunPtr AsapoRequestCallback ->+    Ptr AsapoErrorHandle ->+    IO CInt++foreign import capi "asapo/producer_c.h asapo_producer_send_file"+  asapo_producer_send_file ::+    AsapoProducerHandle ->+    AsapoMessageHeaderHandle ->+    -- file name+    ConstCString ->+    -- ingest mode+    Word64 ->+    -- stream+    ConstCString ->+    FunPtr AsapoRequestCallback ->+    Ptr AsapoErrorHandle ->+    IO CInt++foreign import capi "asapo/producer_c.h asapo_producer_send_stream_finished_flag"+  asapo_producer_send_stream_finished_flag ::+    AsapoProducerHandle ->+    -- stream+    ConstCString ->+    -- last ID+    Word64 ->+    -- next stream+    ConstCString ->+    FunPtr AsapoRequestCallback ->+    Ptr AsapoErrorHandle ->+    IO CInt++foreign import capi "asapo/producer_c.h asapo_producer_send_beamtime_metadata"+  asapo_producer_send_beamtime_metadata ::+    AsapoProducerHandle ->+    -- metadata+    ConstCString ->+    AsapoMetaIngestOp ->+    -- upsert+    AsapoBool ->+    FunPtr AsapoRequestCallback ->+    Ptr AsapoErrorHandle ->+    IO CInt++foreign import capi "asapo/producer_c.h asapo_producer_send_stream_metadata"+  asapo_producer_send_stream_metadata ::+    AsapoProducerHandle ->+    -- metadata+    ConstCString ->+    AsapoMetaIngestOp ->+    -- upsert+    AsapoBool ->+    -- stream+    ConstCString ->+    FunPtr AsapoRequestCallback ->+    Ptr AsapoErrorHandle ->+    IO CInt++foreign import capi "asapo/producer_c.h asapo_request_callback_payload_get_response"+  asapo_request_callback_payload_get_response :: AsapoRequestCallbackPayloadHandle -> IO AsapoStringHandle++foreign import capi "asapo/producer_c.h asapo_request_callback_payload_get_original_header"+  asapo_request_callback_payload_get_original_header :: AsapoRequestCallbackPayloadHandle -> IO (ConstPtr AsapoGenericRequestHeader)++foreign import capi "asapo/producer_c.h asapo_producer_set_log_level"+  asapo_producer_set_log_level :: AsapoProducerHandle -> AsapoLogLevel -> IO ()++foreign import capi "asapo/producer_c.h asapo_producer_enable_local_log"+  asapo_producer_enable_local_log :: AsapoProducerHandle -> AsapoBool -> IO ()++foreign import capi "asapo/producer_c.h asapo_producer_enable_remote_log"+  asapo_producer_enable_remote_log :: AsapoProducerHandle -> AsapoBool -> IO ()++foreign import capi "asapo/producer_c.h asapo_producer_set_credentials"+  asapo_producer_set_credentials ::+    AsapoProducerHandle ->+    AsapoSourceCredentialsHandle ->+    Ptr AsapoErrorHandle ->+    IO CInt++foreign import capi "asapo/producer_c.h asapo_producer_get_requests_queue_size"+  asapo_producer_get_requests_queue_size :: AsapoProducerHandle -> IO Word64++foreign import capi "asapo/producer_c.h asapo_producer_get_requests_queue_volume_mb"+  asapo_producer_get_requests_queue_volume_mb :: AsapoProducerHandle -> IO Word64++foreign import capi "asapo/producer_c.h asapo_producer_set_requests_queue_limits"+  asapo_producer_set_requests_queue_limits ::+    AsapoProducerHandle ->+    -- size+    Word64 ->+    -- volume+    Word64 ->+    IO ()++foreign import capi "asapo/producer_c.h asapo_producer_wait_requests_finished"+  asapo_producer_wait_requests_finished ::+    AsapoProducerHandle ->+    -- timeout ms+    Word64 ->+    -- error+    Ptr AsapoErrorHandle ->+    IO CInt

@@ -0,0 +1,15 @@+module Main where++import Build_doctests (flags, module_sources, pkgs)+-- import Data.Foldable (traverse_)+import System.Environment (unsetEnv)+import Test.DocTest (doctest)++-- This is simply copied from https://github.com/ulidtko/cabal-doctest+main :: IO ()+main = do+  -- traverse_ putStrLn args -- optionally print arguments+  unsetEnv "GHC_ENVIRONMENT" -- see 'Notes'; you may not need this+  doctest args+  where+    args = flags ++ pkgs ++ module_sources