Asapo.Either.Consumer

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.

Documentation

data FilesystemFlag

Whether to use the filesystem or do it in-memory

Constructors

WithFilesystem

WithoutFilesystem

Instances

Eq FilesystemFlag

Defined in Asapo.Either.Consumer

Methods

(==) :: FilesystemFlag -> FilesystemFlag -> Bool

(/=) :: FilesystemFlag -> FilesystemFlag -> Bool

newtype ServerName

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

Constructors

ServerName Text

data IncludeIncompleteFlag

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

Constructors

IncludeIncomplete

ExcludeIncomplete

Instances

Eq IncludeIncompleteFlag

Defined in Asapo.Either.Consumer

Methods

(==) :: IncludeIncompleteFlag -> IncludeIncompleteFlag -> Bool

(/=) :: IncludeIncompleteFlag -> IncludeIncompleteFlag -> Bool

newtype SourcePath

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

Constructors

SourcePath Text

data Dataset

Metadata for a dataset

Constructors

Dataset

Fields

• datasetId :: Word64
• datasetExpectedSize :: Word64
• datasetItems :: [MessageMetaHandle]

data MessageMeta

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

Show MessageMeta

Defined in Asapo.Either.Consumer

Methods

showsPrec :: Int -> MessageMeta -> ShowS

show :: MessageMeta -> String

showList :: [MessageMeta] -> ShowS

data DeleteFlag

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

Constructors

DeleteMeta

DontDeleteMeta

Instances

Eq DeleteFlag

Defined in Asapo.Either.Consumer

Methods

(==) :: DeleteFlag -> DeleteFlag -> Bool

(/=) :: DeleteFlag -> DeleteFlag -> Bool

data ErrorOnNotExistFlag

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

Constructors

ErrorOnNotExist

NoErrorOnNotExist

Instances

Eq ErrorOnNotExistFlag

Defined in Asapo.Either.Consumer

Methods

(==) :: ErrorOnNotExistFlag -> ErrorOnNotExistFlag -> Bool

(/=) :: ErrorOnNotExistFlag -> ErrorOnNotExistFlag -> Bool

data Consumer

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

data NetworkConnectionType

Network connection type

Constructors

ConnectionUndefined

not sure about this

ConnectionTcp

TCP

ConnectionFabric

not sure about this

data StreamFilter

Constructors

FilterAllStreams

FilterFinishedStreams

FilterUnfinishedStreams

data MessageMetaHandle

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

Show MessageMetaHandle

Defined in Asapo.Either.Consumer

Methods

showsPrec :: Int -> MessageMetaHandle -> ShowS

show :: MessageMetaHandle -> String

showList :: [MessageMetaHandle] -> ShowS

data GroupId

Wrapper around a group ID

data Error

Wrapper around an ASAP:O producer error, with an additional error code

Constructors

Error

Fields

• errorMessage :: Text
• errorType :: ErrorType

Instances

Show Error

Defined in Asapo.Either.Consumer

Methods

showsPrec :: Int -> Error -> ShowS

show :: Error -> String

showList :: [Error] -> ShowS

data ErrorType

Constructors

ErrorNoData

ErrorEndOfStream

ErrorStreamFinished

ErrorUnavailableService

ErrorInterruptedTransaction

ErrorLocalIOError

ErrorWrongInput

ErrorPartialData

ErrorUnsupportedClient

ErrorDataNotInCache

ErrorUnknownError

Instances

Show ErrorType

Defined in Asapo.Either.Consumer

Methods

showsPrec :: Int -> ErrorType -> ShowS

show :: ErrorType -> String

showList :: [ErrorType] -> ShowS

withConsumer :: forall a. ServerName -> SourcePath -> FilesystemFlag -> SourceCredentials -> (Error -> IO a) -> (Consumer -> IO a) -> IO a

Create a consumer and do something with it. This is the main entrypoint into the consumer

retrieveDataForMessageMeta :: Consumer -> MessageMeta -> IO (Either Error ByteString)

Retrieve actual data for the handle

queryMessages

Arguments

:: Consumer

-> Text

Actual query string, see the docs for syntax

-> StreamName

-> IO (Either Error [MessageMeta])

Query messages, return handles without data

resendNacs

Arguments

:: Consumer

-> Bool

resend yes/no

-> NominalDiffTime

delay

-> Word64

attempts

-> IO ()

Reset negative acknowledgements

getNextDataset

Arguments

:: Consumer

-> GroupId

-> Word64

Wait until dataset has these number of messages (0 for maximum size)

-> StreamName

-> IO (Either Error Dataset)

Get the next dataset for a stream

retrieveDataFromMeta :: Consumer -> MessageMetaHandle -> IO (Either Error ByteString)

Retrieve data for the given metadata handle

setStreamPersistent :: Consumer -> StreamName -> IO (Either Error Int)

Set a stream persistent

getLastMessageMetaAndData :: Consumer -> StreamName -> IO (Either Error (MessageMeta, ByteString))

Retrieve the last message in the stream, with data and metadata

getLastMessageMeta :: Consumer -> StreamName -> IO (Either Error MessageMeta)

Retrieve the last message in the stream, only metadata (you can get the data later with retrieveDataFromMessageMeta)

getLastMessageData :: Consumer -> StreamName -> IO (Either Error ByteString)

Retrieve the last message in the stream, only data

getLastInGroupMessageMetaAndData :: Consumer -> StreamName -> GroupId -> IO (Either Error (MessageMeta, ByteString))

Retrieve the last message in a given stream and group, with data and metadata

getLastInGroupMessageMeta :: Consumer -> StreamName -> GroupId -> IO (Either Error MessageMeta)

Retrieve the last message in a given stream and group, only metadata (you can get the data later with retrieveDataFromMessageMeta)

getLastInGroupMessageData :: Consumer -> StreamName -> GroupId -> IO (Either Error ByteString)

Retrieve the last message in a given stream and group, only data

getNextMessageMetaAndData :: Consumer -> StreamName -> GroupId -> IO (Either Error (MessageMeta, ByteString))

Retrieve the next message in the stream and group, with data and metadata

getNextMessageMeta :: Consumer -> StreamName -> GroupId -> IO (Either Error MessageMeta)

Retrieve the next message in the stream and group, only metadata (you can get the data later with retrieveDataFromMessageMeta)

getNextMessageData :: Consumer -> StreamName -> GroupId -> IO (Either Error ByteString)

Retrieve the next message in the stream and group, only data

getCurrentDatasetCount :: Consumer -> StreamName -> IncludeIncompleteFlag -> IO (Either Error Int)

Get number of datasets in stream

getBeamtimeMeta :: Consumer -> IO (Either Error (Maybe Text))

Get beamtime metadata (which can be not set, in which case Nothing is returned)

getCurrentSize :: Consumer -> StreamName -> IO (Either Error Int)

Get the current size (number of messages) of the stream

acknowledge :: Consumer -> GroupId -> StreamName -> MessageId -> IO (Either Error Int)

Acknowledge a specific message

negativeAcknowledge

Arguments

:: Consumer

-> GroupId

-> StreamName

-> MessageId

-> NominalDiffTime

delay

-> IO (Either Error Int)

Negatively acknowledge a specific message

getMessageMetaAndDataById :: Consumer -> StreamName -> MessageId -> IO (Either Error (MessageMeta, ByteString))

Given a message ID, retrieve both metadata and data

getMessageMetaById :: Consumer -> StreamName -> MessageId -> IO (Either Error MessageMeta)

Given a message ID, retrieve only the metadata (you can get the data later with retrieveDataFromMessageMeta)

getMessageDataById :: Consumer -> StreamName -> MessageId -> IO (Either Error ByteString)

Given a message ID, retrieve only the data

getUnacknowledgedMessages :: Consumer -> GroupId -> StreamName -> (MessageId, MessageId) -> IO (Either Error [MessageId])

Get a list of all unacknowledged message IDs in a range

withGroupId :: forall a. Consumer -> (Error -> IO a) -> (GroupId -> IO a) -> IO a

Allocate a group ID and call a callback

queryMessagesHandles

Arguments

:: Consumer

-> Text

Actual query string, see the docs for syntax

-> StreamName

-> IO (Either Error [MessageMetaHandle])

Query messages, return handles without data

setTimeout :: Consumer -> NominalDiffTime -> IO ()

Set the global consumer timeout

getLastDataset

Arguments

:: Consumer

-> Word64

Wait until dataset has these number of messages (0 for maximum size)

-> StreamName

-> IO (Either Error Dataset)

Get the last dataset in the stream

getLastDatasetInGroup

Arguments

:: Consumer

-> GroupId

-> Word64

Wait until dataset has these number of messages (0 for maximum size)

-> StreamName

-> IO (Either Error Dataset)

Get the last data ste in the given group

resetLastReadMarker :: Consumer -> GroupId -> StreamName -> IO (Either Error Int)

Reset the last read marker for a specific group

setLastReadMarker :: Consumer -> GroupId -> StreamName -> MessageId -> IO (Either Error Int)

Set the last read marker for the stream

getCurrentConnectionType :: Consumer -> IO NetworkConnectionType

Retrieve the current consumer connection type

getStreamList

Arguments

:: Consumer

-> Maybe StreamName

Pass Nothing to get all streams

-> StreamFilter

-> IO (Either Error [StreamInfo])

Retrieve the list of streams with metadata

deleteStream :: Consumer -> StreamName -> DeleteFlag -> ErrorOnNotExistFlag -> IO (Either Error Int)

Delete a given stream

resolveMetadata :: MessageMetaHandle -> IO MessageMeta

Get the actual metadata hiding behind a handle (shouldn't be necessary when using the convenience interfaces)