Asapo.Producer

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

Types

newtype ProducerException

Constructors

ProducerException Text

Instances

Exception ProducerException

Defined in Asapo.Producer

Methods

toException :: ProducerException -> SomeException

fromException :: SomeException -> Maybe ProducerException

displayException :: ProducerException -> String

Show ProducerException

Defined in Asapo.Producer

Methods

showsPrec :: Int -> ProducerException -> ShowS

show :: ProducerException -> String

showList :: [ProducerException] -> ShowS

newtype Endpoint

Wrapper around an ASAP:O producer endpoint (usually something like "host:port")

Constructors

Endpoint Text

newtype ProcessingThreads

Wrapper around the number of ASAP:O processing threads (simply to make call signatures mor readable)

Constructors

ProcessingThreads Int

data RequestHandlerType

This has no documentation in ASAP:O yet

Constructors

TcpHandler

FilesystemHandler

newtype Error

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.

Constructors

Error Text

Instances

Show Error

Defined in Asapo.Either.Producer

Methods

showsPrec :: Int -> Error -> ShowS

show :: Error -> String

showList :: [Error] -> ShowS

newtype Metadata

Wrapper around metadata to be produced

Constructors

Metadata Text

data SourceCredentials

Constructors

SourceCredentials

Fields

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

newtype MessageId

Constructors

MessageId Word64

Instances

Show MessageId

Defined in Asapo.Either.Common

Methods

showsPrec :: Int -> MessageId -> ShowS

show :: MessageId -> String

showList :: [MessageId] -> ShowS

data DeletionFlags

Constructors

DeleteMeta

Delete metadata also

DeleteErrorOnNotExist

Don't throw an error if the data doesn't exist anyways

Instances

Eq DeletionFlags

Defined in Asapo.Either.Producer

Methods

(==) :: DeletionFlags -> DeletionFlags -> Bool

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

data Producer

Opaque wrapper around an ASAP:O producer

data LogLevel

Constructors

LogNone

LogError

LogInfo

LogDebug

LogWarning

Instances

Eq LogLevel

Defined in Asapo.Either.Producer

Methods

(==) :: LogLevel -> LogLevel -> Bool

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

newtype FileName

Wrapper around file name (dubious to use Text here, but fine for now)

Constructors

FileName Text

newtype PipelineStep

Constructors

PipelineStep Text

newtype Beamtime

Constructors

Beamtime Text

newtype DataSource

Constructors

DataSource Text

newtype Beamline

Constructors

Beamline Text

data StreamInfo

Constructors

StreamInfo

Fields

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

Instances

Show StreamInfo

Defined in Asapo.Either.Common

Methods

showsPrec :: Int -> StreamInfo -> ShowS

show :: StreamInfo -> String

showList :: [StreamInfo] -> ShowS

data SourceType

Constructors

RawSource

ProcessedSource

newtype StreamName

Constructors

StreamName Text

Instances

Show StreamName

Defined in Asapo.Either.Common

Methods

showsPrec :: Int -> StreamName -> ShowS

show :: StreamName -> String

showList :: [StreamName] -> ShowS

newtype InstanceId

Constructors

InstanceId Text

newtype Token

Constructors

Token Text

newtype DatasetSubstream

Wrapper around the substream to use

Constructors

DatasetSubstream Int

newtype DatasetSize

Wrapper around the dataset size to use

Constructors

DatasetSize Int

data VersionInfo

Constructors

VersionInfo

Fields

• versionClient :: Text
• versionServer :: Text
• versionSupported :: Bool

Instances

Show VersionInfo

Defined in Asapo.Either.Producer

Methods

showsPrec :: Int -> VersionInfo -> ShowS

show :: VersionInfo -> String

showList :: [VersionInfo] -> ShowS

data UpsertMode

Constructors

UseUpsert

NoUpsert

data MetadataIngestMode

Constructors

Insert

Replace

Update

data AutoIdFlag

Anti-boolean-blindness for the "auto id" flag in the message header

Constructors

UseAutoId

NoAutoId

Instances

Eq AutoIdFlag

Defined in Asapo.Either.Producer

Methods

(==) :: AutoIdFlag -> AutoIdFlag -> Bool

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

data TransferFlag

Which data to transfer

Constructors

DataAndMetadata

MetadataOnly

data StorageFlag

Where to store the data

Constructors

Filesystem

Database

FilesystemAndDatabase

data RequestResponse

Information about the request and its response, to be used in the ASAP:O send callback

Constructors

RequestResponse

Fields

• responsePayload :: Text
• responseOriginalRequestHeader :: GenericRequestHeader
• responseError :: Maybe Error

data Opcode

Constructors

OpcodeUnknownOp

OpcodeTransferData

OpcodeTransferDatasetData

OpcodeStreamInfo

OpcodeLastStream

OpcodeGetBufferData

OpcodeAuthorize

OpcodeTransferMetaData

OpcodeDeleteStream

OpcodeGetMeta

OpcodeCount

OpcodePersistStream

data GenericRequestHeader

Information about the send request, to be used in the ASAP:O send callback

Constructors

GenericRequestHeader

Fields

• genericRequestHeaderOpCode :: Opcode
• genericRequestHeaderDataId :: Int
• genericRequestHeaderDataSize :: Int
• genericRequestHeaderMetaSize :: Int
• genericRequestHeaderCustomData :: [Int]
• genericRequestHeaderMessage :: ByteString
• genericRequestHeaderStream :: Text
• genericRequestHeaderApiVersion :: Text

Initialization

withProducer

Arguments

:: forall a. Endpoint

-> ProcessingThreads

-> RequestHandlerType

-> SourceCredentials

-> NominalDiffTime

timeout

-> (Producer -> IO a)

-> IO a

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

Getters

getVersionInfo :: Producer -> IO VersionInfo

Retrieve producer version info

getStreamInfo

Arguments

:: Producer

-> StreamName

-> NominalDiffTime

Timeout

-> IO StreamInfo

Retrieve info for a single stream

getLastStream

Arguments

:: Producer

-> NominalDiffTime

Timeout

-> IO StreamInfo

Retrieve info for the latest stream

getBeamtimeMeta

Arguments

:: Producer

-> NominalDiffTime

timeout

-> IO (Maybe Text)

Retrieve metadata for the given stream (which might be missing, in which case Nothing is returned)

getStreamMeta

Arguments

:: Producer

-> StreamName

-> NominalDiffTime

timeout

-> IO (Maybe Text)

Retrieve metadata for the given stream (which might be missing, in which case Nothing is returned)

getRequestsQueueSize :: Producer -> IO Int

Get current size of the requests queue (number of requests pending/being processed)

getRequestsQueueVolumeMb :: Producer -> IO Int

Get current volume of the requests queue (total memory of occupied by pending/being processed requests)

Modifiers

send

Arguments

:: Producer

-> MessageId

-> FileName

-> Metadata

-> DatasetSubstream

-> DatasetSize

-> AutoIdFlag

-> ByteString

Actual data to send

-> TransferFlag

-> StorageFlag

-> StreamName

-> (RequestResponse -> IO ())

-> IO Int

Send a message containing raw data. Due to newtype and enum usage, all parameter should be self-explanatory

sendFile

Arguments

:: Producer

-> MessageId

-> FileName

File name to put into the message header

-> Metadata

-> DatasetSubstream

-> DatasetSize

-> AutoIdFlag

-> Int

Size

-> FileName

File to actually send

-> TransferFlag

-> StorageFlag

-> StreamName

-> (RequestResponse -> IO ())

-> IO Int

Send a message containing a file. Due to newtype and enum usage, all parameter should be self-explanatory

sendStreamFinishedFlag :: Producer -> StreamName -> MessageId -> StreamName -> (RequestResponse -> IO ()) -> IO Int

As the title says, send the "stream finished" flag

sendBeamtimeMetadata :: Producer -> Metadata -> MetadataIngestMode -> UpsertMode -> (RequestResponse -> IO ()) -> IO Int

Send or extend beamtime metadata

sendStreamMetadata :: Producer -> Metadata -> MetadataIngestMode -> UpsertMode -> StreamName -> (RequestResponse -> IO ()) -> IO Int

Send or extend stream metadata

deleteStream

Arguments

:: Producer

-> StreamName

-> NominalDiffTime

timeout

-> [DeletionFlags]

-> IO Int

setLogLevel :: Producer -> LogLevel -> IO ()

Set the log level

enableLocalLog :: Producer -> Bool -> IO ()

Enable/Disable logging to stdout

enableRemoteLog :: Producer -> Bool -> IO ()

Enable/Disable logging to the central server

setCredentials :: Producer -> SourceCredentials -> IO Int

Set a different set of credentials

setRequestsQueueLimits

Arguments

:: Producer

-> Int

Size (0 for unlimited)

-> Int

Volume (in MiB; 0 for unlimited)

-> IO ()

Set maximum size of the requests queue

messageIdFromInt :: Integral a => a -> MessageId

waitRequestsFinished :: Producer -> NominalDiffTime -> IO Int

Wait for all outstanding requests to finish