Documentation
¶
Overview ¶
Package dispatcher implements transport.Dispatcher and transport.SubscriptionSource against a function-table of registered handlers. Contributors register their intent handlers via Register / RegisterSubscription / RegisterContributor; the HTTP and SSE transports look them up at request time.
See SLICE_C_DESIGN.md in the parent contract directory for the spec this implements.
Index ¶
- func RegisterCommand[I, O any](d *Dispatcher, contributor, intent string, version int, ...) error
- func RegisterQuery[I, O any](d *Dispatcher, contributor, intent string, version int, ...) error
- func RegisterSubscription[P, E any](d *Dispatcher, contributor, intent string, version int, ...) error
- type Contributor
- type Dispatcher
- func (d *Dispatcher) Dispatch(ctx context.Context, req contract.Request, p contract.Principal) (json.RawMessage, contract.ResponseMeta, error)
- func (d *Dispatcher) Register(contributor, intent string, version int, h Handler) error
- func (d *Dispatcher) RegisterContributor(c Contributor) error
- func (d *Dispatcher) RegisterSubscription(contributor, intent string, version int, h SubscriptionHandler) error
- func (d *Dispatcher) SetRemoteDispatcher(rd RemoteDispatcher)
- func (d *Dispatcher) Subscribe(ctx context.Context, p contract.Principal, contributor string, ...) (<-chan contract.StreamEvent, func(), error)
- type Handler
- type IdempotencyCached
- type IdempotencyStore
- type IntentRef
- type LoggerAuditEmitter
- type MetricsEmitter
- type NoopMetricsEmitter
- type Option
- type PrometheusMetricsEmitter
- type RemoteDispatcher
- type Result
- type SubscriptionHandler
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func RegisterCommand ¶
func RegisterCommand[I, O any](d *Dispatcher, contributor, intent string, version int, fn func(ctx context.Context, in I, p contract.Principal) (O, error)) error
RegisterCommand is identical in shape to RegisterQuery; both register a query/command handler. The dispatcher's wire layer enforces kind/capability matching against the manifest, so the only practical difference between the two helpers is intent of the caller — they're aliases.
func RegisterQuery ¶
func RegisterQuery[I, O any](d *Dispatcher, contributor, intent string, version int, fn func(ctx context.Context, in I, p contract.Principal) (O, error)) error
RegisterQuery wraps a typed handler in a Handler-compatible closure that JSON-decodes Payload into I and encodes the returned O into Result.Data. I and O must be JSON-marshallable. Use struct{} for an empty-input intent.
func RegisterSubscription ¶
func RegisterSubscription[P, E any](d *Dispatcher, contributor, intent string, version int, fn func(ctx context.Context, in P, p contract.Principal) (<-chan E, func(), error)) error
RegisterSubscription wraps a typed subscription handler. The pump goroutine JSON-encodes each typed E event into a contract.StreamEvent before forwarding into the broker's channel.
Types ¶
type Contributor ¶
type Contributor interface {
Name() string
Handlers() map[IntentRef]Handler
Subscriptions() map[IntentRef]SubscriptionHandler
}
Contributor is layer (b)'s registration shape: a contributor publishes its handler and subscription tables, and the dispatcher walks them on Register.
type Dispatcher ¶
type Dispatcher struct {
// contains filtered or unexported fields
}
Dispatcher is the concrete implementation of transport.Dispatcher and transport.SubscriptionSource (Subscribe lives in subscription.go). Contributors register handlers indexed by (contributor, intent, version); dispatch is a map lookup + the handler call wrapped in metrics emission and canonical error mapping.
func New ¶
func New(metrics MetricsEmitter) *Dispatcher
New returns a fresh dispatcher. Pass NoopMetricsEmitter{} for tests / dev; slice (b) provides a Prometheus-backed implementation.
func NewWithOptions ¶
func NewWithOptions(metrics MetricsEmitter, opts ...Option) *Dispatcher
NewWithOptions returns a dispatcher configured with the supplied options. The existing New(metrics) constructor is preserved as a thin wrapper.
func (*Dispatcher) Dispatch ¶
func (d *Dispatcher) Dispatch(ctx context.Context, req contract.Request, p contract.Principal) (json.RawMessage, contract.ResponseMeta, error)
Dispatch implements transport.Dispatcher. When a tracer is configured, a span wraps the dispatch with attributes capturing (contributor, intent, version, kind) and a status reflecting the outcome.
func (*Dispatcher) Register ¶
func (d *Dispatcher) Register(contributor, intent string, version int, h Handler) error
Register binds a query/command handler to a (contributor, intent, version) key. Returns an error on duplicate registration.
func (*Dispatcher) RegisterContributor ¶
func (d *Dispatcher) RegisterContributor(c Contributor) error
RegisterContributor walks a Contributor's Handlers() and Subscriptions() maps and registers each one. Atomic: if any registration fails, all preceding registrations from this call are rolled back.
func (*Dispatcher) RegisterSubscription ¶
func (d *Dispatcher) RegisterSubscription(contributor, intent string, version int, h SubscriptionHandler) error
RegisterSubscription binds a subscription handler to (contributor, intent, version).
func (*Dispatcher) SetRemoteDispatcher ¶
func (d *Dispatcher) SetRemoteDispatcher(rd RemoteDispatcher)
SetRemoteDispatcher installs (or clears, with nil) the fallback consulted when no local handler matches a request. Idempotent — the dispatcher's forwarding plumbing typically calls this once during wire-up.
func (*Dispatcher) Subscribe ¶
func (d *Dispatcher) Subscribe(ctx context.Context, p contract.Principal, contributor string, intent contract.Intent, params map[string]contract.ParamSource) (<-chan contract.StreamEvent, func(), error)
Subscribe implements transport.SubscriptionSource. The broker calls this on each subscribe-control message; the dispatcher routes to the registered handler. Params from YAML (map[string]contract.ParamSource) are flattened into a runtime map[string]any using the From string when set, the literal Value otherwise.
type Handler ¶
type Handler func(ctx context.Context, payload json.RawMessage, params map[string]any, p contract.Principal) (*Result, error)
Handler is the foundation function-table handler signature for query and command intents. Returning a *contract.Error propagates the canonical code to the wire; any other error is wrapped as CodeInternal at dispatch time.
type IdempotencyCached ¶
type IdempotencyCached struct {
Status int
WireBody json.RawMessage
StoredAt time.Time
TTL time.Duration
}
IdempotencyCached mirrors idempotency.Cached; defined here for the same import-cycle reason. Adapters in the wire-up convert between the two.
type IdempotencyStore ¶
type IdempotencyStore interface {
Lookup(ctx context.Context, key, identity string) (*IdempotencyCached, bool)
Store(ctx context.Context, key, identity string, c IdempotencyCached) error
}
IdempotencyStore is the minimal surface the dispatcher needs from extensions/dashboard/contract/idempotency. Defining it here avoids an import cycle (the idempotency package is consumed only via this interface).
type LoggerAuditEmitter ¶
type LoggerAuditEmitter struct {
// contains filtered or unexported fields
}
LoggerAuditEmitter writes audit records as info-level structured logs via a forge.Logger. Each AuditRecord field is emitted as a discrete log field so log aggregators can filter by `audit=true` cheaply. Pass a nil logger to disable — the emitter then becomes a noop.
func NewLoggerAuditEmitter ¶
func NewLoggerAuditEmitter(logger forge.Logger) *LoggerAuditEmitter
NewLoggerAuditEmitter returns an emitter that writes via logger. Pass nil to disable (the emitter becomes a noop).
func (*LoggerAuditEmitter) Emit ¶
func (e *LoggerAuditEmitter) Emit(_ context.Context, rec contract.AuditRecord)
Emit implements contract.AuditEmitter.
type MetricsEmitter ¶
type MetricsEmitter interface {
RecordDispatch(ctx context.Context, contributor, intent string, version int, kind contract.Kind, latency time.Duration, errCode contract.ErrorCode)
}
MetricsEmitter ships dispatch metrics to a backend. The Phase 4 expansion of this file adds the full DispatchInfo struct and the noop default. For Phase 1, only the interface and the noop are needed.
type NoopMetricsEmitter ¶
type NoopMetricsEmitter struct{}
NoopMetricsEmitter discards all dispatch metrics.
type Option ¶
type Option func(*Dispatcher)
Option configures a Dispatcher.
func WithIdempotencyStore ¶
func WithIdempotencyStore(s IdempotencyStore) Option
WithIdempotencyStore wires command dedup. When set, commands carrying a non-empty IdempotencyKey are deduped per-user via the store.
func WithTracer ¶
WithTracer configures the dispatcher to open a span per Dispatch call. Passing a nil tracer is equivalent to not supplying the option at all.
type PrometheusMetricsEmitter ¶
type PrometheusMetricsEmitter struct {
// contains filtered or unexported fields
}
PrometheusMetricsEmitter records dispatch metrics into a forge.Metrics registry. Counters and histograms are created lazily on first emission (forge.Metrics.Counter / .Histogram are get-or-create). Pass nil to disable — the emitter then becomes a noop.
func NewPrometheusMetricsEmitter ¶
func NewPrometheusMetricsEmitter(m forge.Metrics) *PrometheusMetricsEmitter
NewPrometheusMetricsEmitter returns an emitter that writes to m. If m is nil, the emitter is a noop.
type RemoteDispatcher ¶
type RemoteDispatcher interface {
Dispatch(ctx context.Context, req contract.Request, p contract.Principal) (json.RawMessage, contract.ResponseMeta, error)
}
RemoteDispatcher is the fallback Dispatcher consults when no local handler is registered for an envelope. The dispatcher passes the verbatim request through; implementations typically POST it to a peer service. Slice (m) added this so dashboards can aggregate contributors from multiple upstreams without baking forwarding into the dispatcher itself.
type Result ¶
type Result struct {
// Data is the JSON-encoded response body. May be nil for a {data: null} response.
Data json.RawMessage
// ExtraInvalidates is appended to the manifest's declared Invalidates.
ExtraInvalidates []string
// CacheOverride, when non-nil, replaces the manifest's declared cache hint.
CacheOverride *contract.CacheHint
}
Result carries the data payload plus optional response-meta overrides. Handlers that don't need to influence meta can return &Result{Data: ...}; handlers that need to add invalidations or override cache hints set the extra fields.
type SubscriptionHandler ¶
type SubscriptionHandler func(ctx context.Context, params map[string]any, p contract.Principal) (<-chan contract.StreamEvent, func(), error)
SubscriptionHandler is the function-table handler for subscription intents. The handler returns a channel of events, a force-stop function, and an optional error. Closing the channel signals end-of-stream; cancelling ctx is the canonical way to ask the handler to stop emitting.
Source Files