messenger

package module

Go to main page

Versions in this module

v0

v0.3.1

Oct 3, 2026

v0.3.0

Sep 5, 2026

Changes in this version

+ const DefaultBatchMaxBytes

+ const DefaultBatchMaxMessages

+ const DefaultBatchMaxWait

+ var ErrInvalidBatchResult = errors.New("messenger: invalid batch result")

+ func DeferAfter(err error, delay time.Duration) error

+ func DeferDelay(err error) (time.Duration, bool)

+ type BatchConfig struct

+ MaxBytes int

+ MaxMessages int

+ MaxWait time.Duration

+ Middlewares []BatchMiddleware

+ func (c BatchConfig) Normalize(concurrency int) (BatchConfig, error)

+ type BatchHandler func(context.Context, []Message[T]) (BatchResult, error)

+ func ChainBatchHandler[T any](handler BatchHandler[T], middlewares ...BatchHandlerMiddleware[T]) BatchHandler[T]

+ type BatchHandlerFunc func(context.Context) (BatchResult, error)

+ type BatchHandlerMiddleware func(BatchHandler[T]) BatchHandler[T]

+ type BatchItemKey struct

+ MessageID MessageID

+ Source string

+ type BatchItemResult struct

+ Err error

+ Key BatchItemKey

+ type BatchMiddleware func(ctx context.Context, metadata []Metadata, handlerID string, ...) (BatchResult, error)

+ type BatchPublisher interface

+ PublishBatch func(ctx context.Context, payloads []T) ([]Receipt, error)

+ PublishMessageBatch func(ctx context.Context, outgoing []Outgoing[T]) ([]Receipt, error)

+ func BindBatchPublisher[T any](messenger *Messenger, descriptor Event[T]) BatchPublisher[T]

+ type BatchResult struct

+ Items []BatchItemResult

+ type BatchResultBuilder struct

+ func NewBatchResultBuilder[T any](messages []Message[T]) *BatchResultBuilder[T]

+ func (b *BatchResultBuilder[T]) Build() (BatchResult, error)

+ func (b *BatchResultBuilder[T]) Error(message Message[T]) (error, bool)

+ func (b *BatchResultBuilder[T]) ErrorKey(key BatchItemKey) (error, bool)

+ func (b *BatchResultBuilder[T]) Fail(message Message[T], err error) *BatchResultBuilder[T]

+ func (b *BatchResultBuilder[T]) FailKey(key BatchItemKey, err error) *BatchResultBuilder[T]

+ func (b *BatchResultBuilder[T]) HasErrors() bool

+ func (b *BatchResultBuilder[T]) OK(message Message[T]) *BatchResultBuilder[T]

+ func (b *BatchResultBuilder[T]) OKKey(key BatchItemKey) *BatchResultBuilder[T]

+ type BatchRoute interface

+ DeliverBatch func(ctx context.Context, deliveries []Delivery) ([]Receipt, error)

+ type BatchSender interface

+ SendBatch func(ctx context.Context, payloads []T) ([]Receipt, error)

+ SendMessageBatch func(ctx context.Context, outgoing []Outgoing[T]) ([]Receipt, error)

+ func BindBatchSender[T any](messenger *Messenger, descriptor Command[T]) BatchSender[T]

type Messenger

+ func (m *Messenger) PublishBatch[T any](ctx context.Context, descriptor Event[T], payloads []T) ([]Receipt, error)

+ func (m *Messenger) PublishMessageBatch[T any](ctx context.Context, descriptor Event[T], outgoing []Outgoing[T]) ([]Receipt, error)

+ func (m *Messenger) SendBatch[T any](ctx context.Context, descriptor Command[T], payloads []T) ([]Receipt, error)

+ func (m *Messenger) SendMessageBatch[T any](ctx context.Context, descriptor Command[T], outgoing []Outgoing[T]) ([]Receipt, error)

type Observation

+ BatchACKs int

+ BatchBytes int

+ BatchDLQs int

+ BatchDeferrals int

+ BatchFillDuration time.Duration

+ BatchHandlerDuration time.Duration

+ BatchHandlerMessages int

+ BatchRetries int

+ BatchSize int

type Operation

+ const OperationBatchHandle

+ const OperationExpire

v0.2.2

Aug 29, 2026

v0.2.1

Aug 26, 2026

v0.2.0

Aug 26, 2026

Changes in this version

+ var ErrMessageExpired = errors.New("messenger: message expired")

+ var ErrMessageNotReady = errors.New("messenger: message not ready")

+ var ErrUnsupportedCapability = errors.New("messenger: unsupported route capability")

+ func BoundedFailureText(sanitizer FailureSanitizer, err error, limit int) string

+ func HandlerCompletionError(ctx context.Context, handlerErr error) error

+ func ReportHandlerPanic(ctx context.Context, reporter PanicReporter, handlerID string, recovered any, ...) error

+ func SanitizeError(sanitizer FailureSanitizer, err error) error

+ func SanitizeFailure(sanitizer FailureSanitizer, err error) string

+ type DeepHealthChecker interface

+ DeepHealth func(ctx context.Context) error

+ type FailureSanitizer interface

+ SanitizeFailure func(err error) string

+ func DefaultFailureSanitizer() FailureSanitizer

+ type FailureSanitizerFunc func(error) string

+ func (f FailureSanitizerFunc) SanitizeFailure(err error) string

+ type HandlerPanicError interface

+ HandlerPanicID func() string

+ type LivenessChecker interface

+ Liveness func(ctx context.Context) error

type Observer

+ func NewSanitizedLoggingObserver(logger Logger, sanitizer FailureSanitizer) Observer

type Operation

+ const OperationBrokerAck

+ const OperationDLQHandoff

+ const OperationOffsetCommit

+ const OperationRetryHandoff

type Option

+ func WithPanicReporter(reporter PanicReporter) Option

+ func WithRuntimeShutdownTimeout(timeout time.Duration) Option

+ type PanicReport struct

+ HandlerID string

+ Stack []byte

+ Value any

+ type PanicReporter interface

+ ReportPanic func(ctx context.Context, handlerID string, recovered any, stack []byte)

+ type PanicReporterFunc func(context.Context, PanicReport)

+ func (f PanicReporterFunc) ReportPanic(ctx context.Context, handlerID string, recovered any, stack []byte)

type Runtime

+ func (r *Runtime) DeepHealth(ctx context.Context) error

+ func (r *Runtime) Liveness(ctx context.Context) error

v0.1.0

Aug 25, 2026

Changes in this version

+ const DefaultMaxEnvelopeBytes

+ const DefaultMaxHeaderBytes

+ const DefaultMaxHeaders

+ const EnvelopeSpecVersion

+ const ManifestSpecVersion

+ var ErrDescriptorConflict = errors.New("messenger: descriptor conflict")

+ var ErrEnvelopeTooLarge = errors.New("messenger: envelope too large")

+ var ErrHandlerConflict = errors.New("messenger: handler conflict")

+ var ErrHandlerNotFound = errors.New("messenger: handler not found")

+ var ErrInvalidDescriptor = errors.New("messenger: invalid descriptor")

+ var ErrInvalidMessage = errors.New("messenger: invalid message")

+ var ErrQueryResultMissing = errors.New("messenger: query result missing")

+ var ErrRouteConflict = errors.New("messenger: route conflict")

+ var ErrRouteNotFound = errors.New("messenger: route not found")

+ var ErrRuntimeClosed = errors.New("messenger: runtime closed")

+ var ErrRuntimeNotRunning = errors.New("messenger: runtime not running")

+ var ErrRuntimeRunning = errors.New("messenger: runtime already running")

+ var ErrServiceConflict = errors.New("messenger: service conflict")

+ func CanonicalizeEnvelope(data []byte) ([]byte, error)

+ func ContextWithMetadata(ctx context.Context, metadata Metadata) context.Context

+ func DecodeCommandPayload[T any](descriptor Command[T], data []byte) (T, error)

+ func DecodeEventPayload[T any](descriptor Event[T], data []byte) (T, error)

+ func EncodeCommandEnvelope[T any](descriptor Command[T], metadata Metadata, payload T) ([]byte, error)

+ func EncodeEventEnvelope[T any](descriptor Event[T], metadata Metadata, payload T) ([]byte, error)

+ func EnvelopeFingerprint(data []byte) [sha256.Size]byte

+ func IsPermanent(err error) bool

+ func MarshalEnvelope(metadata Metadata, payload []byte, encoding DataEncoding) ([]byte, error)

+ func Permanent(err error) error

+ func RetryAfter(err error, delay time.Duration) error

+ func RetryDelay(err error) (time.Duration, bool)

+ type Builder struct

+ func NewBuilder(options ...Option) *Builder

+ func (b *Builder) Build() (*Messenger, *Runtime, error)

+ func (b *Builder) HandleCommandFunc[T any](descriptor Command[T], handlerID string, handler PayloadHandler[T])

+ func (b *Builder) HandleCommand[T any](descriptor Command[T], handlerID string, handler Handler[T])

+ func (b *Builder) HandleQueryFunc[Q, R any](descriptor Query[Q, R], handlerID string, handler QueryPayloadHandler[Q, R])

+ func (b *Builder) HandleQuery[Q, R any](descriptor Query[Q, R], handlerID string, handler QueryHandler[Q, R])

+ func (b *Builder) RouteCommand[T any](descriptor Command[T], route Route)

+ func (b *Builder) RouteEvent[T any](descriptor Event[T], route Route)

+ func (b *Builder) RouteQuery[Q, R any](descriptor Query[Q, R], route LocalQueryRoute)

+ func (b *Builder) SubscribeFunc[T any](descriptor Event[T], subscriptionID string, handler PayloadHandler[T])

+ func (b *Builder) Subscribe[T any](descriptor Event[T], subscriptionID string, handler Handler[T])

+ func (b *Builder) Use(serviceID string, service Service)

+ func (b *Builder) UseMiddleware(middlewares ...Middleware)

+ type Codec interface

+ ContentType func() string

+ Decode func(data []byte) (T, error)

+ Encode func(value T) ([]byte, error)

+ Encoding func() DataEncoding

+ func Bytes() Codec[[]byte]

+ func CustomCodec[T any](contentType string, encoding DataEncoding, encode func(T) ([]byte, error), ...) (Codec[T], error)

+ func JSON[T any]() Codec[T]

+ func Text() Codec[string]

+ type Command struct

+ func MustCommand[T any](name string, schemaVersion int, codec Codec[T]) Command[T]

+ func NewCommand[T any](name string, schemaVersion int, codec Codec[T]) (Command[T], error)

+ func (d Command[T]) Info() DescriptorInfo

+ func (d Command[T]) WithSchema(schema string) Command[T]

+ type ContextPropagator interface

+ Extract func(ctx context.Context, carrier map[string]string) context.Context

+ Inject func(ctx context.Context, carrier map[string]string)

+ func NoopContextPropagator() ContextPropagator

+ type DataEncoding uint8

+ const DataBinary

+ const DataJSON

+ const DataText

+ func (e *DataEncoding) UnmarshalJSON(data []byte) error

+ func (e DataEncoding) MarshalJSON() ([]byte, error)

+ func (e DataEncoding) String() string

+ type Delivery interface

+ Fingerprint func() ([sha256.Size]byte, error)

+ HandlerCount func() int

+ Invoke func(ctx context.Context) error

+ MarshalEnvelope func() ([]byte, error)

+ Metadata func() Metadata

+ type DescriptorInfo struct

+ ContentType string

+ DataEncoding DataEncoding

+ Kind Kind

+ Name string

+ Schema string

+ SchemaVersion int

+ type Envelope struct

+ CausationID MessageID

+ ContentType string

+ CorrelationID MessageID

+ Data json.RawMessage

+ DataBase64 *string

+ DataEncoding DataEncoding

+ ExpiresAt time.Time

+ Headers map[string]string

+ ID MessageID

+ Key string

+ Kind Kind

+ Name string

+ NotBefore time.Time

+ Schema string

+ SchemaVersion int

+ Source string

+ SpecVersion string

+ Subject string

+ Time time.Time

+ func UnmarshalEnvelope(data []byte) (Envelope, error)

+ func (e Envelope) Metadata() Metadata

+ func (e Envelope) Payload() ([]byte, DataEncoding, error)

+ func (e Envelope) Validate() error

+ type Event struct

+ func MustEvent[T any](name string, schemaVersion int, codec Codec[T]) Event[T]

+ func NewEvent[T any](name string, schemaVersion int, codec Codec[T]) (Event[T], error)

+ func (d Event[T]) Info() DescriptorInfo

+ func (d Event[T]) WithSchema(schema string) Event[T]

+ type Handler func(context.Context, Message[T]) error

+ func ChainHandler[T any](handler Handler[T], middlewares ...HandlerMiddleware[T]) Handler[T]

+ func HandlePayload[T any](handler PayloadHandler[T]) Handler[T]

+ type HandlerFunc func(context.Context) error

+ type HandlerMiddleware func(Handler[T]) Handler[T]

+ type IDGenerator interface

+ New func() (MessageID, error)

+ func UUIDv7Generator() IDGenerator

+ type Kind string

+ const KindCommand

+ const KindEvent

+ const KindQuery

+ type LocalAsyncConfig struct

+ Capacity int

+ DetachExecution bool

+ Workers int

+ type LocalAsyncRoute struct

+ func NewLocalAsyncRoute(name string, config LocalAsyncConfig) (*LocalAsyncRoute, error)

+ func (r *LocalAsyncRoute) BeginDrain()

+ func (r *LocalAsyncRoute) Deliver(ctx context.Context, delivery Delivery) (Receipt, error)

+ func (r *LocalAsyncRoute) ManagedService() (string, Service)

+ func (r *LocalAsyncRoute) Name() string

+ func (r *LocalAsyncRoute) Readiness(context.Context) error

+ func (r *LocalAsyncRoute) Run(ctx context.Context) error

+ func (r *LocalAsyncRoute) Shutdown(ctx context.Context) error

+ type LocalQueryRoute interface

+ Name func() string

+ type LocalSyncRoute struct

+ func NewLocalSyncRoute() *LocalSyncRoute

+ func (*LocalSyncRoute) Name() string

+ func (r *LocalSyncRoute) Deliver(ctx context.Context, delivery Delivery) (Receipt, error)

+ type LogAttr struct

+ Key string

+ Value any

+ type LogLevel uint8

+ const LogDebug

+ const LogError

+ const LogInfo

+ const LogWarn

+ type Logger interface

+ Log func(ctx context.Context, level LogLevel, message string, attrs ...LogAttr)

+ func AdaptSlog(logger *slog.Logger) Logger

+ type Manifest struct

+ Descriptors []ManifestDescriptor

+ Services []string

+ Source string

+ SpecVersion string

+ func (m Manifest) Validate() error

+ type ManifestDescriptor struct

+ HandlerIDs []string

+ Route string

+ type Message struct

+ Metadata Metadata

+ Payload T

+ func DecodeCommand[T any](descriptor Command[T], data []byte) (Message[T], error)

+ func DecodeEvent[T any](descriptor Event[T], data []byte) (Message[T], error)

+ type MessageID [16]byte

+ func ParseMessageID(value string) (MessageID, error)

+ func (id *MessageID) UnmarshalJSON(data []byte) error

+ func (id *MessageID) UnmarshalText(text []byte) error

+ func (id MessageID) IsZero() bool

+ func (id MessageID) MarshalJSON() ([]byte, error)

+ func (id MessageID) MarshalText() ([]byte, error)

+ func (id MessageID) String() string

+ type Messenger struct

+ func (m *Messenger) Manifest() Manifest

+ func (m *Messenger) MarshalManifest() ([]byte, error)

+ func (m *Messenger) PublishMessage[T any](ctx context.Context, descriptor Event[T], outgoing Outgoing[T]) (Receipt, error)

+ func (m *Messenger) Publish[T any](ctx context.Context, descriptor Event[T], payload T) (Receipt, error)

+ func (m *Messenger) Query[Q, R any](ctx context.Context, descriptor Query[Q, R], payload Q) (R, error)

+ func (m *Messenger) SendMessage[T any](ctx context.Context, descriptor Command[T], outgoing Outgoing[T]) (Receipt, error)

+ func (m *Messenger) Send[T any](ctx context.Context, descriptor Command[T], payload T) (Receipt, error)

+ type Metadata struct

+ CausationID MessageID

+ ContentType string

+ CorrelationID MessageID

+ ExpiresAt time.Time

+ Headers map[string]string

+ ID MessageID

+ Key string

+ Kind Kind

+ Name string

+ NotBefore time.Time

+ Schema string

+ SchemaVersion int

+ Source string

+ Subject string

+ Time time.Time

+ func MetadataFromContext(ctx context.Context) (Metadata, bool)

+ type Middleware func(ctx context.Context, metadata Metadata, handlerID string, next HandlerFunc) error

+ type Observation struct

+ Attempt uint64

+ ConsumerID string

+ Duplicate bool

+ Duration time.Duration

+ Err error

+ HandlerID string

+ Kind Kind

+ MessageID MessageID

+ Name string

+ Operation Operation

+ RetryDelay time.Duration

+ Route string

+ SchemaVersion int

+ ServiceID string

+ StartedAt time.Time

+ State ReceiptState

+ type Observer interface

+ Observe func(ctx context.Context, observation Observation)

+ func NewLoggingObserver(logger Logger) Observer

+ type Operation string

+ const OperationDeliver

+ const OperationHandle

+ const OperationQuery

+ const OperationService

+ type Option func(*Builder)

+ func WithClock(clock func() time.Time) Option

+ func WithContextPropagator(propagator ContextPropagator) Option

+ func WithIDGenerator(generator IDGenerator) Option

+ func WithLogger(logger Logger) Option

+ func WithObserver(observer Observer) Option

+ func WithSource(source string) Option

+ type Outgoing struct

+ Metadata OutgoingMetadata

+ Payload T

+ type OutgoingMetadata struct

+ CausationID MessageID

+ CorrelationID MessageID

+ ExpiresAt time.Time

+ Headers map[string]string

+ ID MessageID

+ Key string

+ NotBefore time.Time

+ Subject string

+ Time time.Time

+ type PayloadHandler func(context.Context, T) error

+ type Publisher interface

+ Publish func(ctx context.Context, payload T) (Receipt, error)

+ PublishMessage func(ctx context.Context, outgoing Outgoing[T]) (Receipt, error)

+ func BindPublisher[T any](messenger *Messenger, descriptor Event[T]) Publisher[T]

+ type Querier interface

+ Query func(ctx context.Context, payload Q) (R, error)

+ func BindQuerier[Q, R any](messenger *Messenger, descriptor Query[Q, R]) Querier[Q, R]

+ type Query struct

+ func MustQuery[Q, R any](name string, schemaVersion int, codec Codec[Q]) Query[Q, R]

+ func NewQuery[Q, R any](name string, schemaVersion int, codec Codec[Q]) (Query[Q, R], error)

+ func (d Query[Q, R]) Info() DescriptorInfo

+ func (d Query[Q, R]) WithSchema(schema string) Query[Q, R]

+ type QueryHandler func(context.Context, Message[Q]) (R, error)

+ func ChainQueryHandler[Q, R any](handler QueryHandler[Q, R], middlewares ...QueryHandlerMiddleware[Q, R]) QueryHandler[Q, R]

+ func HandleQueryPayload[Q, R any](handler QueryPayloadHandler[Q, R]) QueryHandler[Q, R]

+ type QueryHandlerMiddleware func(QueryHandler[Q, R]) QueryHandler[Q, R]

+ type QueryPayloadHandler func(context.Context, Q) (R, error)

+ type Receipt struct

+ At time.Time

+ MessageID MessageID

+ Route string

+ State ReceiptState

+ type ReceiptState string

+ const ReceiptAccepted

+ const ReceiptBrokerConfirmed

+ const ReceiptCompleted

+ const ReceiptNoop

+ const ReceiptStaged

+ type Route interface

+ Deliver func(ctx context.Context, delivery Delivery) (Receipt, error)

+ Name func() string

+ type Runtime struct

+ func (r *Runtime) BeginDrain()

+ func (r *Runtime) Readiness(ctx context.Context) error

+ func (r *Runtime) Run(ctx context.Context) error

+ func (r *Runtime) Shutdown(ctx context.Context) error

+ type Sender interface

+ Send func(ctx context.Context, payload T) (Receipt, error)

+ SendMessage func(ctx context.Context, outgoing Outgoing[T]) (Receipt, error)

+ func BindSender[T any](messenger *Messenger, descriptor Command[T]) Sender[T]

+ type Service interface

+ BeginDrain func()

+ Readiness func(ctx context.Context) error

+ Run func(ctx context.Context) error

+ Shutdown func(ctx context.Context) error

+ type ServiceProvider interface

+ ManagedService func() (serviceID string, service Service)

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL