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
- Variables
- func IDOf(msg hermod.Message) (string, bool)
- func ModeOf(config map[string]string) (wait bool, timeout time.Duration)
- func NewConversationID() string
- func Resolve(id string, o Outcome) bool
- func Take(msg hermod.Message) (string, bool)
- func ValidConversationID(id string) bool
- type Outcome
- type Pending
- type Status
Constants ¶
const ( HeaderConversationID = "X-Conversation-Id" MetaConversationID = "conversation_id" )
A conversation id ties the messages of one chat together, so a workflow can remember what was said earlier in it. A caller sends it with each message (HeaderConversationID) and the transport puts it in the message's metadata (MetaConversationID); a synchronous caller that sent none is given one in the answer, to send with its next message.
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 NewConversationID ¶ added in v1.18.0
func NewConversationID() string
NewConversationID starts a conversation.
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.
func ValidConversationID ¶ added in v1.18.0
ValidConversationID reports whether id is 1 to 128 characters of letters, digits, '.', '_', ':' and '-'. The id comes from the caller and ends up in storage keys and response headers, so anything else is refused rather than escaped.
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