Documentation
¶
Index ¶
- func IncomingTraceIDFromContext(ctx context.Context) string
- func TraceIDFromContext(ctx context.Context) string
- func WithConsumerGroup(ctx context.Context, group string) context.Context
- type ExpvarMetricsRecorder
- type Inbox
- type Logging
- type Metrics
- type NoopMetricsRecorder
- type Recover
- type Tracing
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func IncomingTraceIDFromContext ¶
IncomingTraceIDFromContext returns the trace ID supplied by the cross-process producer when strict mode is in effect; in the default (trusting) mode this returns "" because the producer-supplied value is already used as the active TraceIDFromContext.
func TraceIDFromContext ¶
TraceIDFromContext returns the trace ID stamped onto ctx by the Tracing consume middleware. In the default (trusting) mode this is the same value the producer sent; in strict mode it is a freshly generated, non-spoofable ID.
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) 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) 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) 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) 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 a trace identifier across the publish/consume boundary. On publish, a missing TraceID is filled with a fresh UUID and stashed into Headers under the W3C-compatible "traceparent" key. On consume, the TraceID is restored onto the context so downstream loggers can attach it to every record.
Two modes are supported:
- Default (NewTracing): incoming TraceIDs 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 TraceIDs are treated as untrusted. A fresh ID is generated for log correlation and the declared incoming value is parked under a separate ctx key accessible via IncomingTraceIDFromContext. 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 TraceIDs (W3C-compatible propagation).
func NewTracingStrict ¶
func NewTracingStrict() *Tracing
NewTracingStrict constructs a Tracing middleware that does not trust incoming TraceIDs. 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) 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 injects a trace ID into the envelope.
Source Files