Documentation
¶
Overview ¶
Package wire holds the message formats shared by the engine and the SDKs beyond commands and events: worker jobs, dead letters and cache entries, plus the publish helpers that apply the claim-check and deduplication ids.
Index ¶
- Constants
- func CmdID(execID, key string, gen, attempt int) string
- func JobMsgID(execID, key string, gen, attempt int) string
- func PublishCmd(ctx context.Context, in *infra.Infra, c cmd.Command, traceparent string) error
- func PublishDLQ(ctx context.Context, in *infra.Infra, d DeadLetter, dedup string) error
- func TriggerStart(flow, execID string, data []byte) (cmd.Command, error)
- type CacheEntry
- type Concurrency
- type DeadLetter
- type Job
- type Progress
- type Reply
- type Stop
- type StopTask
Constants ¶
const ( HeaderMsgID = "Nats-Msg-Id" HeaderTraceparent = "traceparent" HeaderProtocol = "Pt-Protocol-Version" )
Header names.
const ( ReplyInvalid = "invalid" ReplyRejected = "rejected" ReplyNotFound = "not_found" ReplyFailed = "failed" )
Reply codes.
const ( DLQCommand = "cmd" DLQJob = "job" DLQTrigger = "trigger" // DLQDispatch records an event the dispatcher gave up on (its execution // is quarantined). DLQDispatch = "dispatch" )
Dead-letter kinds.
Variables ¶
This section is empty.
Functions ¶
func CmdID ¶
CmdID derives the id of the command a worker sends for one attempt: a redelivered job produces the same id, so the stream deduplicates it.
func PublishCmd ¶
PublishCmd publishes a command to its execution's partition, applying the claim-check and using the command id as the deduplication id (see msgID).
func PublishDLQ ¶
PublishDLQ records a dead letter. It is deduplicated per (kind, key, original subject sequence) so a crash between recording and terminating the original message does not record it twice.
Types ¶
type CacheEntry ¶
type CacheEntry struct {
Output json.RawMessage `json:"output,omitempty"`
Writes map[string]json.RawMessage `json:"writes,omitempty"`
Next string `json:"next,omitempty"`
Usage map[string]float64 `json:"usage,omitempty"`
Expires time.Time `json:"expires"`
}
CacheEntry is a cached task result.
type Concurrency ¶
Concurrency is a per-key concurrency limit a worker must honour.
type DeadLetter ¶
type DeadLetter struct {
Kind string `json:"kind"`
Key string `json:"key"`
// Flow is the flow a trigger dead letter was meant to start.
Flow string `json:"flow,omitempty"`
Reason string `json:"reason"`
Deliveries uint64 `json:"deliveries,omitempty"`
Time time.Time `json:"time"`
Subject string `json:"subject"`
Header map[string][]string `json:"header,omitempty"`
Body []byte `json:"body,omitempty"`
// Seq is the dead-letter stream sequence (set on read).
Seq uint64 `json:"-"`
}
DeadLetter is a message the system gave up on.
type Job ¶
type Job struct {
ExecID string `json:"exec_id"`
Flow string `json:"flow"`
FlowHash string `json:"flow_hash"`
Node string `json:"node"`
Kind string `json:"kind"`
Key string `json:"key"`
Index int `json:"index,omitempty"`
Generation int `json:"generation"`
Attempt int `json:"attempt"`
Context json.RawMessage `json:"context"`
// Meta is the node's meta from the execution's flow version.
Meta json.RawMessage `json:"meta,omitempty"`
// Reply is the command subject the worker publishes its result to.
Reply string `json:"reply"`
CacheKey string `json:"cache_key,omitempty"`
Concurrency *Concurrency `json:"concurrency,omitempty"`
Traceparent string `json:"traceparent,omitempty"`
}
Job is a unit of work for a worker of Kind.
type Progress ¶
type Progress struct {
ExecID string `json:"exec_id"`
Node string `json:"node"`
Key string `json:"key"`
Generation int `json:"generation"`
Attempt int `json:"attempt"`
Seq int64 `json:"seq"`
Time time.Time `json:"time"`
Data json.RawMessage `json:"data,omitempty"`
}
Progress is an intermediate result a worker publishes while it runs a job: core NATS, never written to the log, received only by clients already listening (G5-04). Seq counts the messages of one job attempt from 1.
type Reply ¶
type Reply struct {
OK bool `json:"ok"`
Seq uint64 `json:"seq,omitempty"`
Code string `json:"code,omitempty"`
Error string `json:"error,omitempty"`
}
Reply is the engine's answer to a command that carries a reply subject (cmd.Command.Reply), sent once the command is decided: OK with the sequence of the decision it produced (or of the last one, for a no-op), or a Code: "invalid" (malformed or does not fit the flow), "rejected" (refused in the execution's current state), "not_found", "failed" (dead-lettered).
type Stop ¶
type Stop struct {
ExecID string `json:"exec_id"`
All bool `json:"all,omitempty"`
Tasks []StopTask `json:"tasks,omitempty"`
}
Stop tells workers that running jobs of an execution are no longer wanted: every job of the execution (All: it ended) or the listed attempts (a task was cancelled, an attempt timed out). Published on core NATS by the dispatcher, best effort: a worker that misses it learns from the log head that the execution ended, and any late result is stale anyway (G5-02).
Source Files