Documentation
¶
Overview ¶
Package reply carries what a workflow did with a message back to a caller that is waiting for it.
A push source answers its caller as soon as the message is queued, so the caller never learns whether the workflow delivered it. A transport that wants to answer with the result instead calls Expect before it dispatches the message and waits; the engine, when it has finished with the message, calls Resolve. The two meet through an id carried in the message's metadata, so nothing between them — source wrappers, buffers, the router — has to know.
Both ends are in one process: a push source's transport and the engine that runs its workflow already have to be.
Index ¶
Constants ¶
const ( ConfigMode = "response_mode" ConfigTimeout = "response_timeout" // ModeSync holds the caller until the workflow has finished with the // message. Anything else, including nothing, answers once it is queued. ModeSync = "sync" )
Config keys a push source carries to say how it answers its caller.
const ( // DefaultTimeout is how long a synchronous caller is held when the source // names no timeout, or one that does not parse. DefaultTimeout = 30 * time.Second // MaxTimeout caps what a source may ask for. A request held open is a // connection and a waiter held open. MaxTimeout = 5 * time.Minute )
const MetaReplyID = "_hermod_reply_id"
MetaReplyID is the metadata key under which an awaited message carries the id its waiter is registered under.
Variables ¶
var ErrTooManyPending = errors.New("too many requests are waiting for a workflow result")
ErrTooManyPending is returned by Expect when the limit on callers waiting at once is reached.
Functions ¶
func ModeOf ¶
ModeOf reads how a source answers from its configuration: whether the caller waits, and for how long.
A timeout that is missing, unreadable or not positive is the default, never "no timeout" and never a fall back to answering at once.
func Resolve ¶
Resolve hands an outcome to the caller waiting under id, and reports whether there was one. A message is resolved once: the waiter is removed as it is answered, so a replayed message carrying the same id reaches nobody.
func Take ¶
Take returns the reply id of an awaited message and removes it from the message. The engine calls it as it starts on a message: the id is how it finds the caller, and from there on it is nobody else's business — left on the message it is written to every sink and handed back in the record.
Types ¶
type Outcome ¶
type Outcome struct {
Status Status
// Error describes the failure. Empty for Delivered and Completed.
Error string
// Record is the message as the workflow left it, as JSON: what a sink was
// sent when it was delivered, and the message itself otherwise.
Record []byte
}
Outcome is what the waiting caller is told.
type Pending ¶
type Pending struct {
// contains filtered or unexported fields
}
Pending is one caller waiting for one message.
func Expect ¶
Expect marks msg as awaited and registers the caller as its waiter. Call it before the message is dispatched, and Cancel when done waiting.
type Status ¶
type Status string
Status says what the workflow did with a message.
const ( // Delivered: written to every sink the workflow routed it to. Delivered Status = "delivered" // Completed: the workflow ran and had nothing to write — it has no sink, or // it is a dry run. Completed Status = "completed" // Filtered: the workflow ran and chose to deliver it nowhere — a filter // dropped it, or its outcome had no edge to follow. Filtered Status = "filtered" // DeadLettered: it failed and was parked in the dead-letter sink. DeadLettered Status = "dead_lettered" // Failed: it failed and is not preserved anywhere. Failed Status = "failed" )
Source Files