Documentation
¶
Overview ¶
Package protocol defines the public NATS request/reply contract between the Packtrail engine and the remote services that implement workflow tasks.
This is the only package in the module intended to be imported by external projects. Everything under internal/ is private to the engine.
Index ¶
Constants ¶
const ( StatusOK = "ok" // task succeeded; Payload is the new shared context StatusError = "error" // task failed permanently (no retry requested) StatusRetry = "retry" // task asks the engine to retry per the node policy )
Task response status values.
Variables ¶
This section is empty.
Functions ¶
func Serve ¶
func Serve(ctx context.Context, nc *nats.Conn, subject string, h Handler) (*nats.Subscription, error)
Serve subscribes a Handler to subject as a queue subscriber (queue group "packtrail-workers"), decoding TaskRequest and replying with TaskResponse. It is a convenience for task services and tests; the engine itself never calls it.
ctx is the worker's lifetime: every handler invocation derives its context from it (tightened by the request's per-call deadline), so cancelling ctx at shutdown cancels in-flight handlers. Cancelling ctx does not unsubscribe — the returned subscription should still be drained/unsubscribed by the caller when done.
subject may contain NATS wildcards (e.g. "tasks.triage.*") so a single worker can serve every execution of a task.
func ServeNamespaced ¶
func ServeNamespaced( ctx context.Context, nc *nats.Conn, namespace, subject string, h Handler, ) (*nats.Subscription, error)
ServeNamespaced is like Serve but prepends namespace to subject, matching the convention used by the engine's built-in nats-task invoker. Use it for out-of-process workers so they subscribe to the right namespaced subject without having to construct it manually.
protocol.ServeNamespaced(ctx, nc, "packtrail", "tasks.triage.*", handler) // subscribes to "packtrail.tasks.triage.*"
Types ¶
type Handler ¶
type Handler func(ctx context.Context, req TaskRequest) (TaskResponse, error)
Handler implements the business logic of a single task. It receives the decoded request and returns a response. Returning a non-nil error is treated as a transient failure and reported to the engine as StatusRetry.
type TaskRequest ¶
type TaskRequest struct {
ExecutionID string `json:"execution_id"`
NodeID string `json:"node_id"`
Payload json.RawMessage `json:"payload"`
Generation uint64 `json:"generation,omitempty"`
Attempt int `json:"attempt"`
Deadline time.Time `json:"deadline"`
}
TaskRequest is the envelope sent by the engine to a task subject via request/reply.
type TaskResponse ¶
type TaskResponse struct {
Status string `json:"status"`
Payload json.RawMessage `json:"payload,omitempty"`
Error string `json:"error,omitempty"`
}
TaskResponse is the envelope a task service returns to the engine.
Source Files