xbus: Core & Client API

Xbus Core & Client APIs

Services

service ActorAgent

ActorAgent must be implemented by receiver actors (workers and consumers)

Process(ActorProcessRequest)

Arguments:

• arg – ActorProcessRequest

Returns:

ActorProcessingState

Process a pipeline step

This is a streamed reply method

If the last reply state is ‘RUNNING’, the agent MUST call ProcessingEnd

service Broker

In charge of recepting the envelopes from any actor.

Output(OutputRequest)

Arguments:

• arg – OutputRequest

Returns:

EnvelopeAck

Called by an actor to send an envelope on a output

service ClientAPI

ClientAPI provides account-based methods

In some case a client may not now all the details of the account it is connected with. This service allows it to get all this information.

GetActors(Void)

Returns:ClientGetActorsReply

GetActors lists the actors of a connected account

service CoreEvents

Internal notification messages

Actor(NoRequest)

Arguments:

• arg – NoRequest

Returns:

CoreEvent

Process(NoRequest)

Arguments:

• arg – NoRequest

Returns:

CoreEvent

ProcessNodeReady(NoRequest)

Arguments:

• arg – NoRequest

Returns:

ProcessNodeReadyEvent

Envelope(NoRequest)

Arguments:

• arg – NoRequest

Returns:

EnvelopeEvent

service Director

Orchestrate the actor processings.

ProcessingEnd(ActorProcessingState)

Arguments:

• arg – ActorProcessingState

Returns:

Void

ProcessingEnd is for signaling the end of a processing by an actor.

If a node closes all its outputs, the signal is done automatically. In all other cases (errors, need to close all outputs but the one used, or consumers), it is required.

Log(ActorLogRequest)

Arguments:

• arg – ActorLogRequest

Returns:

Void

service EnvelopeStorage

Gives access to the stored envelopes

Retrieve(EnvelopePosition)

Arguments:

• arg – EnvelopePosition

Returns:

EnvelopeFragmentAndPosition

Retrieve an envelope from a given position The envelope is returned as a stream of envelope fragments.

service ProcessState

Methods for an emitter to query its process states

EnvelopeStates(NoRequest)

Arguments:

• arg – NoRequest

Returns:

EmitterEnvelopeState

Emits the process states associated to an envelope

service Sentinel

The Sentinel service is in charge of always knowing which actors are online

Survey(NoRequest)

Arguments:

• arg – NoRequest

Returns:

SurveyRequest

All actors should subscribe to this function and act accordingly to the received SurveyRequest

ActorReady(Void)

Returns:NoReply

Once a receiver actor is ready (ie, its ActorAgent implementation is up and subscribed), it must call this function to signal its availability to the bus

ActorLeaving(Void)

Returns:NoReply

Before unsubscribing its ActorAgent implementation, an actor must call this function

ListActors(Void)

Returns:ClientGetActorsReply

Returns the list of actors with their ‘online’ state updated. The reply may be delayed until the list is sure to be uptodate

service SvcRegistration

Registration API

Register(RegistrationRequest)

Arguments:

• arg – RegistrationRequest

Returns:

Registration

Update(RegistrationRequest)

Arguments:

• arg – RegistrationRequest

Returns:

Registration

service WhoAmI

A service for auto-discovering the current identity of the client when only its name or api key is known

WhoAmI(Void)

Returns:Account

Returns the current account details. identifier can be an account name or an api key

Message types

message xbus.Account

Field id: Type id:bytes Field name:Account name Type name:string Field type:Account type Type type:Type Field status:Account status Type status:Status Field csr:The certificate signing request Type csr:string Field csrOrigin:   Type csrOrigin:CSROrigin Field cert:The account certificate Type cert:string Field apiKey:The account api key Type apiKey:string

message xbus.Actor

Field id: Type id:bytes Field name: Type name:string Field kind: Type kind:Kind Field status: Type status:Status Field accountID:   Type accountID:bytes Field roles: Type roles:string Field lastSeen: Type lastSeen:Timestamp Field online: Type online:bool Field unresponsive:   Type unresponsive:  bool Field config: Type config:ConfigEntry

message xbus.ActorLogRequest

Field envelopeID:   Type envelopeID:  bytes Field Context: Type Context:ProcessingContext Field messages: Type messages:LogMessage

message xbus.ActorProcessRequest

Field context: Type context:ProcessingContext Field inputs: Type inputs:Input

message xbus.Input

Field name:Name of the input Type name:string Field close:True if no envelope to expect Type close:bool Field envelope:First envelope fragment if available Type envelope:Envelope Field position:Envelope position Type position:EnvelopePosition

message xbus.ActorProcessingState

Field context: Type context:ProcessingContext Field status: Type status:Status Field messages: Type messages:LogMessage Field ActorLeaving:   Type ActorLeaving:  bool

message xbus.ActorReadyMsg

message xbus.ClientGetActorsReply

Field actors: Type actors:Actor

message xbus.ClientGetActorsRequest

Field actorID: Type actorID:bytes

message xbus.ConfigEntry

Field key: Type key:string Field value: Type value:string

message xbus.CoreEvent

Field name: Type name:string Field actor: Type actor:Actor Field process: Type process:Process Field processOldStatus:   Type processOldStatus:  Status

message xbus.EmitterEnvelopeState

Field processID:   Type processID:bytes Field status: Type status:Status Field errors: Type errors:LogMessage

message xbus.Envelope

Field id: Type id:bytes Field eventIDs: Type eventIDs:bytes Field events: Type events:Event Field last: Type last:bool

message xbus.EnvelopeAck

Field id: Type id:bytes Field status: Type status:ReceptionStatus Field reason: Type reason:string

message xbus.EnvelopeEvent

Field envelopeID:   Type envelopeID:  bytes Field status: Type status:Status Field newData: Type newData:bool

message xbus.EnvelopeFragmentAndPosition

Field fragment: Type fragment:Envelope Field position: Type position:EnvelopePosition

message xbus.EnvelopePosition

Field envelope_id:   Type envelope_id:  bytes Field start: Type start:bool Field complete: Type complete:bool Field eventPositions:   Type eventPositions:  EventPosition

message xbus.EnvelopeTarget

Field actorID: Type actorID:bytes Field nodeID: Type nodeID:string Field input: Type input:string

message xbus.Event

Field id: Type id:bytes Field type: Type type:string Field index: Type index:uint64 Field itemCount:   Type itemCount:uint64 Field checksum: Type checksum:uint32 Field items: Type items:bytes

message xbus.EventPosition

Field event_id: Type event_id:bytes Field index: Type index:uint64 Field itemCount:   Type itemCount:uint64

message xbus.LogEntry

Field envelopeID:   Type envelopeID:  bytes Field actorID: Type actorID:bytes Field processID:   Type processID:bytes Field nodeID: Type nodeID:string Field message: Type message:LogMessage

message xbus.LogMessage

Field time:timestamp of the log entry Type time:Timestamp Field level:severity level Type level:LogLevel Field Text:entry text Type Text:string

message xbus.OutputRequest

Field context: Type context:ProcessingContext Field output: Type output:string Field close: Type close:bool Field envelope: Type envelope:Envelope

message xbus.PMProcess

Field process: Type process:Process Field level: Type level:LogLevel Field status: Type status:Status Field comment: Type comment:string Field logs: Type logs:LogEntry

message xbus.PipelineInfo

Field id: Type id:bytes Field name: Type name:string Field version: Type version:string Field status: Type status:Status

message xbus.Process

Field id: Type id:bytes Field triggerEmitterID:   Type triggerEmitterID:  bytes Field triggerEnvelopeID:   Type triggerEnvelopeID:  bytes Field pipelineID:   Type pipelineID:  bytes Field status: Type status:Status Field state: Type state:string

message xbus.ProcessFilter

Field id: Type id:bytes Field triggerEmitterID:   Type triggerEmitterID:  bytes Field triggerEnvelopeID:   Type triggerEnvelopeID:  bytes Field pipelineID:   Type pipelineID:  bytes Field envelopeID:   Type envelopeID:  bytes Field status: Type status:Status Field statusChangedFrom:   Type statusChangedFrom:  Timestamp Field statusChangedTo:   Type statusChangedTo:  Timestamp

message xbus.ProcessNodeReadyEvent

Field processID:   Type processID:bytes Field actorID: Type actorID:bytes Field nodeID: Type nodeID:string

message xbus.ProcessingContext

Field processID:   Type processID:bytes Field nodeID: Type nodeID:string

message xbus.Registration

Field id: Type id:bytes Field registrationStatus:   Type registrationStatus:  Status Field signedCertificate:   Type signedCertificate:  string Field serverCA: Type serverCA:string Field actorList:   Type actorList:Actor

message xbus.RegistrationRequest

Field name: Type name:string Field type: Type type:Type Field csr: Type csr:string Field actorList:   Type actorList:Actor

message xbus.StringReply

Field value: Type value:string

message xbus.SurveyRequest

Field onlineStatus:   Type onlineStatus:  bool