diff --git a/ChangeLog b/ChangeLog new file mode 100644 --- /dev/null +++ b/ChangeLog @@ -0,0 +1,8 @@ +2024-08-16 Philipp Middendorf + + * Add documentation and high-level bindings + +2024-08-08 Philipp Middendorf + + * Initial work on 0.9.0, just the raw bindings are here + diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -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. diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,6 @@ +module Main where + +import Distribution.Extra.Doctest (defaultMainWithDoctests) + +main :: IO () +main = defaultMainWithDoctests "doctests" diff --git a/app/SimpleConsumer.hs b/app/SimpleConsumer.hs new file mode 100644 --- /dev/null +++ b/app/SimpleConsumer.hs @@ -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 diff --git a/app/SimpleProducer.hs b/app/SimpleProducer.hs new file mode 100644 --- /dev/null +++ b/app/SimpleProducer.hs @@ -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) diff --git a/hs-asapo.cabal b/hs-asapo.cabal new file mode 100644 --- /dev/null +++ b/hs-asapo.cabal @@ -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. ) + * 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 diff --git a/lib/Asapo/Consumer.hs b/lib/Asapo/Consumer.hs new file mode 100644 --- /dev/null +++ b/lib/Asapo/Consumer.hs @@ -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) diff --git a/lib/Asapo/Either/Common.hs b/lib/Asapo/Either/Common.hs new file mode 100644 --- /dev/null +++ b/lib/Asapo/Either/Common.hs @@ -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) diff --git a/lib/Asapo/Either/Consumer.hs b/lib/Asapo/Either/Consumer.hs new file mode 100644 --- /dev/null +++ b/lib/Asapo/Either/Consumer.hs @@ -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) diff --git a/lib/Asapo/Either/Producer.hs b/lib/Asapo/Either/Producer.hs new file mode 100644 --- /dev/null +++ b/lib/Asapo/Either/Producer.hs @@ -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)) diff --git a/lib/Asapo/Producer.hs b/lib/Asapo/Producer.hs new file mode 100644 --- /dev/null +++ b/lib/Asapo/Producer.hs @@ -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) diff --git a/lib/Asapo/Raw/Common.hs b/lib/Asapo/Raw/Common.hs new file mode 100644 --- /dev/null +++ b/lib/Asapo/Raw/Common.hs @@ -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 diff --git a/lib/Asapo/Raw/Consumer.hs b/lib/Asapo/Raw/Consumer.hs new file mode 100644 --- /dev/null +++ b/lib/Asapo/Raw/Consumer.hs @@ -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 diff --git a/lib/Asapo/Raw/FreeHandleHack.hsc b/lib/Asapo/Raw/FreeHandleHack.hsc new file mode 100644 --- /dev/null +++ b/lib/Asapo/Raw/FreeHandleHack.hsc @@ -0,0 +1,19 @@ +{-# LANGUAGE BlockArguments #-} +{-# LANGUAGE CApiFFI #-} +{-# LANGUAGE ForeignFunctionInterface #-} +{-# LANGUAGE GeneralizedNewtypeDeriving #-} + +#include + +-- | +-- 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 ()) diff --git a/lib/Asapo/Raw/Producer.hsc b/lib/Asapo/Raw/Producer.hsc new file mode 100644 --- /dev/null +++ b/lib/Asapo/Raw/Producer.hsc @@ -0,0 +1,425 @@ +{-# LANGUAGE CApiFFI #-} +{-# LANGUAGE BlockArguments #-} +{-# LANGUAGE ForeignFunctionInterface #-} +{-# LANGUAGE GeneralizedNewtypeDeriving #-} + +-- hsc2hs needs the include to find the C structure +#include + +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 diff --git a/tests/doctests.hs b/tests/doctests.hs new file mode 100644 --- /dev/null +++ b/tests/doctests.hs @@ -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