Safe Haskell	Safe-Inferred
Language	Haskell98

Asapo.Consumer

Contents

Error types
Types
Initialization
Getters
Modifiers

Description

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)
:}

Synopsis

data SomeConsumerException
data NoData
data EndOfStream
data StreamFinished
data UnavailableService
data InterruptedTransaction
data LocalIOError
data WrongInput
data PartialData
data UnsupportedClient
data DataNotInCache
data UnknownError
data Consumer
data Dataset = Dataset {
- datasetId :: Word64
- datasetExpectedSize :: Word64
- datasetItems :: [MessageMetaHandle]
}
data MessageMetaHandle
data DeleteFlag
- = DeleteMeta
- | DontDeleteMeta
data ErrorOnNotExistFlag
- = ErrorOnNotExist
- | NoErrorOnNotExist
data ErrorType = ErrorNoData
data FilesystemFlag
- = WithFilesystem
- | WithoutFilesystem
newtype PipelineStep = PipelineStep Text
newtype Beamtime = Beamtime Text
newtype DataSource = DataSource Text
newtype Beamline = Beamline Text
data StreamInfo = StreamInfo {
- streamInfoLastId :: MessageId
- streamInfoName :: StreamName
- streamInfoFinished :: Bool
- streamInfoNextStream :: Text
- streamInfoCreated :: UTCTime
- streamInfoLastEntry :: UTCTime
}
data GroupId
data IncludeIncompleteFlag
- = IncludeIncomplete
- | ExcludeIncomplete
data MessageMeta = MessageMeta {
- messageMetaName :: Text
- messageMetaTimestamp :: UTCTime
- messageMetaSize :: Word64
- messageMetaId :: MessageId
- messageMetaSource :: Text
- messageMetaMetadata :: Text
- messageMetaBufId :: Word64
- messageMetaDatasetSubstream :: Word64
- messageMetaHandle :: MessageMetaHandle
}
newtype ServerName = ServerName Text
newtype SourcePath = SourcePath Text
data SourceType
- = RawSource
- | ProcessedSource
newtype StreamName = StreamName Text
newtype InstanceId = InstanceId Text
newtype Token = Token Text
data StreamFilter
- = FilterAllStreams
- | FilterFinishedStreams
- | FilterUnfinishedStreams
data SourceCredentials = SourceCredentials {
- sourceType :: SourceType
- instanceId :: InstanceId
- pipelineStep :: PipelineStep
- beamtime :: Beamtime
- beamline :: Beamline
- dataSource :: DataSource
- token :: Token
}
data NetworkConnectionType
- = ConnectionUndefined
- | ConnectionTcp
- | ConnectionFabric
withConsumer :: forall a. ServerName -> SourcePath -> FilesystemFlag -> SourceCredentials -> (Consumer -> IO a) -> IO a
withGroupId :: forall a. Consumer -> (GroupId -> IO a) -> IO a
getCurrentSize :: Consumer -> StreamName -> IO Int
getCurrentDatasetCount :: Consumer -> StreamName -> IncludeIncompleteFlag -> IO Int
getBeamtimeMeta :: Consumer -> IO (Maybe Text)
getNextDataset :: Consumer -> GroupId -> Word64 -> StreamName -> IO Dataset
getLastDataset :: Consumer -> Word64 -> StreamName -> IO Dataset
getLastDatasetInGroup :: Consumer -> GroupId -> Word64 -> StreamName -> IO Dataset
getMessageMetaAndDataById :: Consumer -> StreamName -> MessageId -> IO (MessageMeta, ByteString)
getMessageMetaById :: Consumer -> StreamName -> MessageId -> IO MessageMeta
getMessageDataById :: Consumer -> StreamName -> MessageId -> IO ByteString
getNextMessageMetaAndData :: Consumer -> StreamName -> GroupId -> IO (MessageMeta, ByteString)
getNextMessageMeta :: Consumer -> StreamName -> GroupId -> IO MessageMeta
getNextMessageData :: Consumer -> StreamName -> GroupId -> IO ByteString
getLastMessageMetaAndData :: Consumer -> StreamName -> IO (MessageMeta, ByteString)
getLastMessageMeta :: Consumer -> StreamName -> IO MessageMeta
getLastMessageData :: Consumer -> StreamName -> IO ByteString
getLastInGroupMessageMetaAndData :: Consumer -> StreamName -> GroupId -> IO (MessageMeta, ByteString)
getLastInGroupMessageMeta :: Consumer -> StreamName -> GroupId -> IO MessageMeta
getLastInGroupMessageData :: Consumer -> StreamName -> GroupId -> IO ByteString
getUnacknowledgedMessages :: Consumer -> GroupId -> StreamName -> (MessageId, MessageId) -> IO [MessageId]
getCurrentConnectionType :: Consumer -> IO NetworkConnectionType
getStreamList :: Consumer -> Maybe StreamName -> StreamFilter -> IO [StreamInfo]
messageIdFromInt :: Integral a => a -> MessageId
queryMessages :: Consumer -> Text -> StreamName -> IO [MessageMeta]
retrieveDataForMessageMeta :: Consumer -> MessageMeta -> IO ByteString
resetLastReadMarker :: Consumer -> GroupId -> StreamName -> IO Int
setTimeout :: Consumer -> NominalDiffTime -> IO ()
setLastReadMarker :: Consumer -> GroupId -> StreamName -> MessageId -> IO Int
setStreamPersistent :: Consumer -> StreamName -> IO Int
acknowledge :: Consumer -> GroupId -> StreamName -> MessageId -> IO Int
negativeAcknowledge :: Consumer -> GroupId -> StreamName -> MessageId -> NominalDiffTime -> IO Int
deleteStream :: Consumer -> StreamName -> DeleteFlag -> ErrorOnNotExistFlag -> IO Int
resendNacs :: Consumer -> Bool -> NominalDiffTime -> Word64 -> IO ()

Error types

data SomeConsumerException Source #

Parent class for all consumer-related exceptions. This makes catchall possible, as in...

setStreamPersistent consumer (StreamName "default")
  catch (e -> error ("Caught " <> (show (e :: SomeConsumerException))))

Instances

Instances details

Exception SomeConsumerException Source #
Instance details Defined in Asapo.Consumer Methods toException :: SomeConsumerException -> SomeException # fromException :: SomeException -> Maybe SomeConsumerException # displayException :: SomeConsumerException -> String #
Show SomeConsumerException Source #
Instance details Defined in Asapo.Consumer Methods showsPrec :: Int -> SomeConsumerException -> ShowS # show :: SomeConsumerException -> String # showList :: [SomeConsumerException] -> ShowS #

data NoData Source #

Instances

Instances details

Exception NoData Source #
Instance details Defined in Asapo.Consumer Methods toException :: NoData -> SomeException # fromException :: SomeException -> Maybe NoData # displayException :: NoData -> String #
Show NoData Source #
Instance details Defined in Asapo.Consumer Methods showsPrec :: Int -> NoData -> ShowS # show :: NoData -> String # showList :: [NoData] -> ShowS #

data EndOfStream Source #

Instances

Instances details

Exception EndOfStream Source #
Instance details Defined in Asapo.Consumer Methods toException :: EndOfStream -> SomeException # fromException :: SomeException -> Maybe EndOfStream # displayException :: EndOfStream -> String #
Show EndOfStream Source #
Instance details Defined in Asapo.Consumer Methods showsPrec :: Int -> EndOfStream -> ShowS # show :: EndOfStream -> String # showList :: [EndOfStream] -> ShowS #

data StreamFinished Source #

Instances

Instances details

Exception StreamFinished Source #
Instance details Defined in Asapo.Consumer Methods toException :: StreamFinished -> SomeException # fromException :: SomeException -> Maybe StreamFinished # displayException :: StreamFinished -> String #
Show StreamFinished Source #
Instance details Defined in Asapo.Consumer Methods showsPrec :: Int -> StreamFinished -> ShowS # show :: StreamFinished -> String # showList :: [StreamFinished] -> ShowS #

data UnavailableService Source #

Instances

Instances details

Exception UnavailableService Source #
Instance details Defined in Asapo.Consumer Methods toException :: UnavailableService -> SomeException # fromException :: SomeException -> Maybe UnavailableService # displayException :: UnavailableService -> String #
Show UnavailableService Source #
Instance details Defined in Asapo.Consumer Methods showsPrec :: Int -> UnavailableService -> ShowS # show :: UnavailableService -> String # showList :: [UnavailableService] -> ShowS #

data InterruptedTransaction Source #

Instances

Instances details

Exception InterruptedTransaction Source #
Instance details Defined in Asapo.Consumer Methods toException :: InterruptedTransaction -> SomeException # fromException :: SomeException -> Maybe InterruptedTransaction # displayException :: InterruptedTransaction -> String #
Show InterruptedTransaction Source #
Instance details Defined in Asapo.Consumer Methods showsPrec :: Int -> InterruptedTransaction -> ShowS # show :: InterruptedTransaction -> String # showList :: [InterruptedTransaction] -> ShowS #

data LocalIOError Source #

Instances

Instances details

Exception LocalIOError Source #
Instance details Defined in Asapo.Consumer Methods toException :: LocalIOError -> SomeException # fromException :: SomeException -> Maybe LocalIOError # displayException :: LocalIOError -> String #
Show LocalIOError Source #
Instance details Defined in Asapo.Consumer Methods showsPrec :: Int -> LocalIOError -> ShowS # show :: LocalIOError -> String # showList :: [LocalIOError] -> ShowS #

data WrongInput Source #

Instances

Instances details

Exception WrongInput Source #
Instance details Defined in Asapo.Consumer Methods toException :: WrongInput -> SomeException # fromException :: SomeException -> Maybe WrongInput # displayException :: WrongInput -> String #
Show WrongInput Source #
Instance details Defined in Asapo.Consumer Methods showsPrec :: Int -> WrongInput -> ShowS # show :: WrongInput -> String # showList :: [WrongInput] -> ShowS #

data PartialData Source #

Instances

Instances details

Exception PartialData Source #
Instance details Defined in Asapo.Consumer Methods toException :: PartialData -> SomeException # fromException :: SomeException -> Maybe PartialData # displayException :: PartialData -> String #
Show PartialData Source #
Instance details Defined in Asapo.Consumer Methods showsPrec :: Int -> PartialData -> ShowS # show :: PartialData -> String # showList :: [PartialData] -> ShowS #

data UnsupportedClient Source #

Instances

Instances details

Exception UnsupportedClient Source #
Instance details Defined in Asapo.Consumer Methods toException :: UnsupportedClient -> SomeException # fromException :: SomeException -> Maybe UnsupportedClient # displayException :: UnsupportedClient -> String #
Show UnsupportedClient Source #
Instance details Defined in Asapo.Consumer Methods showsPrec :: Int -> UnsupportedClient -> ShowS # show :: UnsupportedClient -> String # showList :: [UnsupportedClient] -> ShowS #

data DataNotInCache Source #

Instances

Instances details

Exception DataNotInCache Source #
Instance details Defined in Asapo.Consumer Methods toException :: DataNotInCache -> SomeException # fromException :: SomeException -> Maybe DataNotInCache # displayException :: DataNotInCache -> String #
Show DataNotInCache Source #
Instance details Defined in Asapo.Consumer Methods showsPrec :: Int -> DataNotInCache -> ShowS # show :: DataNotInCache -> String # showList :: [DataNotInCache] -> ShowS #

data UnknownError Source #

Instances

Instances details

Exception UnknownError Source #
Instance details Defined in Asapo.Consumer Methods toException :: UnknownError -> SomeException # fromException :: SomeException -> Maybe UnknownError # displayException :: UnknownError -> String #
Show UnknownError Source #
Instance details Defined in Asapo.Consumer Methods showsPrec :: Int -> UnknownError -> ShowS # show :: UnknownError -> String # showList :: [UnknownError] -> ShowS #

Types

data Consumer Source #

Wrapper around a consumer handle. Create with the withConsumer function(s).

data Dataset Source #

Metadata for a dataset

Constructors

Dataset
Fields datasetId :: Word64 datasetExpectedSize :: Word64 datasetItems :: [MessageMetaHandle]

data MessageMetaHandle Source #

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)

Instances

Instances details

Show MessageMetaHandle Source #
Instance details Defined in Asapo.Either.Consumer Methods showsPrec :: Int -> MessageMetaHandle -> ShowS # show :: MessageMetaHandle -> String # showList :: [MessageMetaHandle] -> ShowS #

data DeleteFlag Source #

Anti-boolean-blindness for delete or not delete metadata when deleting a stream

Constructors

DeleteMeta
DontDeleteMeta

Instances

Instances details

Eq DeleteFlag Source #
Instance details Defined in Asapo.Either.Consumer Methods (==) :: DeleteFlag -> DeleteFlag -> Bool # (/=) :: DeleteFlag -> DeleteFlag -> Bool #

data ErrorOnNotExistFlag Source #

Anti-boolean-blindness for "error on not existing data"

Constructors

ErrorOnNotExist
NoErrorOnNotExist

Instances

Instances details

Eq ErrorOnNotExistFlag Source #
Instance details Defined in Asapo.Either.Consumer Methods (==) :: ErrorOnNotExistFlag -> ErrorOnNotExistFlag -> Bool # (/=) :: ErrorOnNotExistFlag -> ErrorOnNotExistFlag -> Bool #

data ErrorType Source #

Constructors

ErrorNoData

Instances

Instances details

Show ErrorType Source #
Instance details Defined in Asapo.Either.Consumer Methods showsPrec :: Int -> ErrorType -> ShowS # show :: ErrorType -> String # showList :: [ErrorType] -> ShowS #

data FilesystemFlag Source #

Whether to use the filesystem or do it in-memory

Constructors

WithFilesystem
WithoutFilesystem

Instances

Instances details

Eq FilesystemFlag Source #
Instance details Defined in Asapo.Either.Consumer Methods (==) :: FilesystemFlag -> FilesystemFlag -> Bool # (/=) :: FilesystemFlag -> FilesystemFlag -> Bool #

newtype PipelineStep Source #

Constructors

PipelineStep Text

newtype Beamtime Source #

Constructors

Beamtime Text

newtype DataSource Source #

Constructors

DataSource Text

newtype Beamline Source #

Constructors

Beamline Text

data StreamInfo Source #

Constructors

StreamInfo
Fields streamInfoLastId :: MessageId streamInfoName :: StreamName streamInfoFinished :: Bool streamInfoNextStream :: Text streamInfoCreated :: UTCTime streamInfoLastEntry :: UTCTime

Instances

Instances details

Show StreamInfo Source #
Instance details Defined in Asapo.Either.Common Methods showsPrec :: Int -> StreamInfo -> ShowS # show :: StreamInfo -> String # showList :: [StreamInfo] -> ShowS #

data GroupId Source #

Wrapper around a group ID

data IncludeIncompleteFlag Source #

Anti-boolean-blindness for "include incomplete data sets in list"

Constructors

IncludeIncomplete
ExcludeIncomplete

Instances

Instances details

Eq IncludeIncompleteFlag Source #
Instance details Defined in Asapo.Either.Consumer Methods (==) :: IncludeIncompleteFlag -> IncludeIncompleteFlag -> Bool # (/=) :: IncludeIncompleteFlag -> IncludeIncompleteFlag -> Bool #

data MessageMeta Source #

Metadata for a single message

Constructors

MessageMeta
Fields messageMetaName :: Text messageMetaTimestamp :: UTCTime messageMetaSize :: Word64 messageMetaId :: MessageId messageMetaSource :: Text messageMetaMetadata :: Text messageMetaBufId :: Word64 messageMetaDatasetSubstream :: Word64 messageMetaHandle :: MessageMetaHandle

Instances

Instances details

Show MessageMeta Source #
Instance details Defined in Asapo.Either.Consumer Methods showsPrec :: Int -> MessageMeta -> ShowS # show :: MessageMeta -> String # showList :: [MessageMeta] -> ShowS #

newtype ServerName Source #

Wrapper for a server name (something like "host:port")

Constructors

ServerName Text

newtype SourcePath Source #

Wrapper for a source path (dubious to not use FilePath, but let's see)

Constructors

SourcePath Text

data SourceType Source #

Constructors

RawSource
ProcessedSource

newtype StreamName Source #

Constructors

StreamName Text

Instances

Instances details

Show StreamName Source #
Instance details Defined in Asapo.Either.Common Methods showsPrec :: Int -> StreamName -> ShowS # show :: StreamName -> String # showList :: [StreamName] -> ShowS #

newtype InstanceId Source #

Constructors

InstanceId Text

newtype Token Source #

Constructors

Token Text

data StreamFilter Source #

Constructors

FilterAllStreams
FilterFinishedStreams
FilterUnfinishedStreams

data SourceCredentials Source #

Constructors

SourceCredentials
Fields sourceType :: SourceType instanceId :: InstanceId pipelineStep :: PipelineStep beamtime :: Beamtime beamline :: Beamline dataSource :: DataSource token :: Token