Documentation
¶
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func WithConsumerGroup ¶
WithConsumerGroup stores the consumer group name on ctx so the inbox middleware can scope its dedupe key. The framework Bus is responsible for populating this value before invoking the consume pipeline.
Types ¶
type ExpvarMetricsRecorder ¶
type ExpvarMetricsRecorder struct {
// contains filtered or unexported fields
}
ExpvarMetricsRecorder publishes counts and total latency under stable expvar names: vef_event_publish_count, vef_event_publish_error, vef_event_consume_count, vef_event_consume_error, vef_event_consume_latency_ms. Suitable for default observability via /debug/vars or an expvar-to-Prometheus collector.
func NewExpvarMetricsRecorder ¶
func NewExpvarMetricsRecorder() *ExpvarMetricsRecorder
NewExpvarMetricsRecorder constructs the framework's default recorder. expvar.Publish panics on duplicate names so existing maps are reused when constructed more than once (e.g. across test re-runs).
func (*ExpvarMetricsRecorder) ConsumeObserved ¶
func (r *ExpvarMetricsRecorder) ConsumeObserved(eventType string, elapsed time.Duration, err error)
ConsumeObserved implements event.MetricsRecorder.
func (*ExpvarMetricsRecorder) PublishObserved ¶
func (r *ExpvarMetricsRecorder) PublishObserved(eventType string, err error)
PublishObserved implements event.MetricsRecorder.
type Inbox ¶
type Inbox struct {
// contains filtered or unexported fields
}
Inbox is a consume-side ConsumeMiddleware that dedupes deliveries by (consumer_group, event_id). It activates only on transports whose Capabilities advertise AtLeastOnce so in-memory paths don't pay the database round-trip cost. Handlers must still make their business side effects idempotent: if processing exceeds the lease and the transport redelivers, another worker may acquire and run the same event.
func NewInbox ¶
func NewInbox(repo inbox.Repository, processingLease time.Duration) *Inbox
NewInbox constructs an Inbox middleware.
func (*Inbox) Applies ¶
func (*Inbox) Applies(caps transport.Capabilities) bool
Applies attaches the middleware only when the transport may deliver the same message more than once.
func (*Inbox) Order ¶ added in v0.28.0
Order implements ConsumeMiddleware. Inbox runs innermost so the dedupe decision sits closest to the handler's side effects, and so its panic-time lock release unwinds before the outer Recover middleware converts the panic into an error.
func (*Inbox) WrapConsume ¶
func (m *Inbox) WrapConsume(next middleware.ConsumeHandler) middleware.ConsumeHandler
WrapConsume claims the (group, eventID) slot; on duplicate it short-circuits the handler chain with success so the transport Acks the message and skips redelivery downstream.
type Logging ¶
type Logging struct {
// contains filtered or unexported fields
}
Logging logs publish and consume events at debug/info level. Errors are escalated to warn so they show up in default operational logs.
func NewLogging ¶
NewLogging constructs a Logging middleware.
func (*Logging) Applies ¶
func (*Logging) Applies(transport.Capabilities) bool
Applies always returns true; logging is cross-cutting.
func (*Logging) Order ¶ added in v0.28.0
Order implements both middleware interfaces. Logging runs after tracing so log lines can carry the trace ID.
func (*Logging) WrapConsume ¶
func (m *Logging) WrapConsume(next middleware.ConsumeHandler) middleware.ConsumeHandler
WrapConsume implements ConsumeMiddleware.
func (*Logging) WrapPublish ¶
func (m *Logging) WrapPublish(next middleware.PublishHandler) middleware.PublishHandler
WrapPublish implements PublishMiddleware.
type Metrics ¶
type Metrics struct {
// contains filtered or unexported fields
}
Metrics is a thin pass-through middleware that delegates every observation to a pluggable event.MetricsRecorder. The default expvar-backed recorder is installed by the fx module; applications can decorate the recorder to forward observations to Prometheus, OpenTelemetry, or their own sink.
func NewMetrics ¶
func NewMetrics(rec event.MetricsRecorder) *Metrics
NewMetrics constructs a Metrics middleware that forwards observations to the supplied recorder. A nil recorder is replaced with a no-op so the middleware is always safe to attach.
func (*Metrics) Applies ¶
func (*Metrics) Applies(transport.Capabilities) bool
Applies always returns true.
func (*Metrics) Order ¶ added in v0.28.0
Order implements both middleware interfaces. Metrics measures the handler plus the inner middlewares.
func (*Metrics) WrapConsume ¶
func (m *Metrics) WrapConsume(next middleware.ConsumeHandler) middleware.ConsumeHandler
WrapConsume reports every consume outcome plus elapsed time through the recorder.
func (*Metrics) WrapPublish ¶
func (m *Metrics) WrapPublish(next middleware.PublishHandler) middleware.PublishHandler
WrapPublish reports every publish outcome through the recorder.
type NoopMetricsRecorder ¶
type NoopMetricsRecorder struct{}
NoopMetricsRecorder satisfies event.MetricsRecorder but discards every observation. Useful as a fallback when an application disables metrics entirely while leaving the middleware enabled.
func (*NoopMetricsRecorder) ConsumeObserved ¶
func (*NoopMetricsRecorder) ConsumeObserved(string, time.Duration, error)
ConsumeObserved implements event.MetricsRecorder.
func (*NoopMetricsRecorder) PublishObserved ¶
func (*NoopMetricsRecorder) PublishObserved(string, error)
PublishObserved implements event.MetricsRecorder.
type Recover ¶
type Recover struct {
// contains filtered or unexported fields
}
Recover is a ConsumeMiddleware that converts handler panics into errors so the consume pipeline can apply standard retry / DLQ logic. The stack trace is logged at error level for postmortem.
func NewRecover ¶
NewRecover constructs a Recover middleware.
func (*Recover) Applies ¶
func (*Recover) Applies(transport.Capabilities) bool
Applies implements ConsumeMiddleware. Recover is always useful so it attaches to every transport.
func (*Recover) Order ¶ added in v0.28.0
Order implements ConsumeMiddleware. Recover wraps outermost so it captures panics from every inner middleware and the handler.
func (*Recover) WrapConsume ¶
func (m *Recover) WrapConsume(next middleware.ConsumeHandler) middleware.ConsumeHandler
WrapConsume implements ConsumeMiddleware.
type Tracing ¶
type Tracing struct {
// contains filtered or unexported fields
}
Tracing propagates W3C trace context across the publish/consume boundary. On publish it ensures the envelope carries a 32-hex trace ID and a fresh 16-hex span ID, and stamps a valid `traceparent` header (version 00). On consume it restores the trace ID onto the context so downstream loggers and handlers can correlate every record.
Two modes are supported:
- Default (NewTracing): incoming trace IDs are trusted and propagated end-to-end, matching W3C / OpenTelemetry conventions. This is the correct choice for receiving events inside a trust boundary (intra-cluster RPC, internal pub/sub).
- Strict (NewTracingStrict): incoming trace IDs are treated as untrusted. A fresh ID is generated for log correlation, the declared incoming value is parked under a separate ctx key accessible via IncomingTraceIDFromContext, and the envelope handed to the handler carries the fresh ID — never the untrusted one. Use when accepting events from less-trusted producers (multi-tenant ingress, public webhook → outbox bridge).
func NewTracing ¶
func NewTracing() *Tracing
NewTracing constructs a Tracing middleware that trusts incoming trace IDs (W3C-compatible propagation).
func NewTracingStrict ¶
func NewTracingStrict() *Tracing
NewTracingStrict constructs a Tracing middleware that does not trust incoming trace IDs. A fresh trace ID is generated per consume and the declared incoming value is exposed only through IncomingTraceIDFromContext so audit pipelines can distinguish forged from framework-generated IDs.
func (*Tracing) Applies ¶
func (*Tracing) Applies(transport.Capabilities) bool
Applies always returns true; tracing is cross-cutting.
func (*Tracing) Order ¶ added in v0.28.0
Order implements both middleware interfaces. Tracing runs early so the trace context it installs is visible to logging, metrics, and the handler.
func (*Tracing) WrapConsume ¶
func (m *Tracing) WrapConsume(next middleware.ConsumeHandler) middleware.ConsumeHandler
WrapConsume restores the trace ID into the context. Behavior depends on the strict flag — see the Tracing type documentation.
func (*Tracing) WrapPublish ¶
func (*Tracing) WrapPublish(next middleware.PublishHandler) middleware.PublishHandler
WrapPublish ensures the envelope carries a valid W3C trace context and stamps the traceparent header.
Source Files