messaging

package module
v0.3.2 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 23, 2026 License: MIT Imports: 14 Imported by: 0

README

go-messaging

Transport-neutral messaging contracts and independently versioned transport modules for Go. The root module owns immutable envelopes, routing, ingress, delivery dispositions, capabilities, correlation, and observability. Provider SDKs stay in nested modules.

Modules

Module Purpose
github.com/goliatone/go-messaging Contracts, envelopes, routes, ingress, and reply correlation
github.com/goliatone/go-messaging/transport/valkey Valkey Pub/Sub and Streams drivers
github.com/goliatone/go-messaging/adapters/go-command Optional typed bridge for go-command v0.23.1+
github.com/goliatone/go-messaging/adapters/go-admin Optional go-admin Command Runs bridge and Valkey Pub/Sub assembly

Install only the modules used by the application:

go get github.com/goliatone/go-messaging@latest
go get github.com/goliatone/go-messaging/transport/valkey@latest
go get github.com/goliatone/go-messaging/adapters/go-command@latest
go get github.com/goliatone/go-messaging/adapters/go-admin@latest

Basic publisher

driver, err := pubsub.New(pubsub.DefaultConfig("127.0.0.1:6379"))
if err != nil {
    return err
}
if err := driver.Start(ctx); err != nil {
    return err
}
defer driver.Close(context.Background())

drivers, err := messaging.NewDriverRegistry(map[string]messaging.Driver{
    "valkey-pubsub": driver,
})
if err != nil {
    return err
}
router, err := messaging.NewRouter(drivers, []messaging.Route{{
    Name:     "domain-events",
    Strategy: messaging.StrategyPrimary,
    Kinds:    []messaging.Kind{messaging.KindEvent},
    Bindings: []messaging.RouteBinding{{
        Driver:      "valkey-pubsub",
        Destination: messaging.Destination{Name: "events"},
    }},
}}, nil)
if err != nil {
    return err
}

event := messaging.NewEnvelope(
    "event-42", "user.created", messaging.KindEvent,
    "1", "application/json", payload, nil,
)
_, err = router.Publish(ctx, "domain-events", event)

Use Valkey Streams when durable delivery, acknowledgement, competing consumers, replay, or dead-lettering is required. Use Pub/Sub for ephemeral fanout. Routes are logical application names; provider destinations remain inside route bindings.

Structured errors

Messaging keeps its sentinel and typed errors compatible with the standard errors.Is and errors.As APIs. At logging, API, or adapter boundaries, project an error into the shared go-errors contract without changing the original:

if structured := messaging.AsGoError(err); structured != nil {
    logger.LogAttrs(ctx, slog.LevelError, "publish failed", messaging.ErrorSlogAttributes(err)...)
}

if retryable := messaging.AsRetryableError(err); retryable != nil {
    delay := retryable.RetryDelay(attempt)
    // Apply the application's retry policy.
}

Retryability is conservative: a definitely-not-published operation may be retried, while an ambiguous publication is never automatically retryable. A non-nil AsRetryableError result always reports IsRetryable() == true; disabled or critical-severity retryable values produce nil. Structured logging includes only bounded classification and transport metadata; provider causes and payloads remain excluded.

Runnable chat example

examples/chat-demo provides a complete browser and terminal chat over a go-router WebSocket gateway and Valkey Pub/Sub. It is a nested module, so its UI and transport dependencies do not enter the root module.

go-command integration

The optional adapter supports local-only, publisher, worker, hybrid placement, explicitly authorized event-to-command triggers, and correlated command/query replies. See the go-command adapter guide for assembly examples and safety rules.

go-admin Command Runs integration

The separately versioned go-admin adapter maps the go-admin command-run contract to strict messaging envelopes. Its optional Valkey package assembles publisher-only workers, subscriber-only web gateways, and hybrid processes on one application/environment Pub/Sub channel. Every web gateway owns an independent subscription; scope validation remains mandatory, driver lifecycle remains host-owned, and Pub/Sub provides no replay.

Development and releases

Run every present module independently:

./taskfile go:race
./taskfile go:vet

Preview a synchronized release without changing files, refs, or remotes:

./taskfile release:dry-run 0.2.0

Before an actual release, ./taskfile release:preflight verifies the branch and tracked tree, rejects stale untracked .version/CHANGELOG.md outputs, and checks git-cliff plus the origin remote. The dry-run validates semantic import-version rules, lists only present nested-module tags, preserves the reviewed go-command requirement, and prints the atomic push refspec.

Documentation

Index

Constants

View Source 
const (
	TextCodeInvalidEnvelope           = "INVALID_ENVELOPE"
	TextCodeSchemaMismatch            = "SCHEMA_MISMATCH"
	TextCodeMessageTooLarge           = "MESSAGE_TOO_LARGE"
	TextCodeUnknownRoute              = "UNKNOWN_ROUTE"
	TextCodeUnknownDriver             = "UNKNOWN_DRIVER"
	TextCodeUnsupportedCapability     = "UNSUPPORTED_CAPABILITY"
	TextCodePublishRejected           = "PUBLISH_REJECTED"
	TextCodePublishAmbiguous          = "PUBLISH_AMBIGUOUS"
	TextCodeNotPublished              = "NOT_PUBLISHED"
	TextCodeSubscriptionNotReady      = "SUBSCRIPTION_NOT_READY"
	TextCodeSubscriptionClosed        = "SUBSCRIPTION_CLOSED"
	TextCodeReplyTimeout              = "REPLY_TIMEOUT"
	TextCodeCorrelationFailure        = "CORRELATION_FAILURE"
	TextCodeAcknowledgementFailed     = "ACKNOWLEDGEMENT_FAILED"
	TextCodeUnsupportedDisposition    = "UNSUPPORTED_DISPOSITION"
	TextCodeDeadLetterFailed          = "DEAD_LETTER_FAILED"
	TextCodeHandlerPanic              = "HANDLER_PANIC"
	TextCodeMessageHandlingFailed     = "MESSAGE_HANDLING_FAILED"
	TextCodeObservationFailed         = "OBSERVATION_FAILED"
	TextCodeOperationCanceled         = "OPERATION_CANCELED"
	TextCodeOperationDeadlineExceeded = "OPERATION_DEADLINE_EXCEEDED"
	TextCodeInternalError             = "MESSAGING_INTERNAL_ERROR"
)
View Source 
const JSONEnvelopeContentType = "application/vnd.goliatone.messaging-envelope+json"
View Source 
const ReservedHeaderPrefix = "messaging."

Variables

View Source 
var (
	ErrInvalidEnvelope = errors.New("messaging: invalid envelope")
	ErrSchemaMismatch  = errors.New("messaging: schema mismatch")
	ErrMessageTooLarge = errors.New("messaging: message too large")
)
View Source 
var (
	ErrUnknownRoute           = errors.New("messaging: unknown route")
	ErrUnknownDriver          = errors.New("messaging: unknown driver")
	ErrUnsupportedCapability  = errors.New("messaging: unsupported capability")
	ErrPublishRejected        = errors.New("messaging: publish rejected")
	ErrPublishAmbiguous       = errors.New("messaging: publish outcome ambiguous")
	ErrNotPublished           = errors.New("messaging: definitely not published")
	ErrSubscriptionNotReady   = errors.New("messaging: subscription not ready")
	ErrSubscriptionClosed     = errors.New("messaging: subscription closed")
	ErrReplyTimeout           = errors.New("messaging: reply timeout")
	ErrCorrelation            = errors.New("messaging: reply correlation failure")
	ErrAcknowledgement        = errors.New("messaging: acknowledgement failure")
	ErrUnsupportedDisposition = errors.New("messaging: unsupported disposition")
	ErrDeadLetter             = errors.New("messaging: dead-letter failure")
	ErrHandlerPanic           = errors.New("messaging: handler panic")
	ErrMessageHandling        = errors.New("messaging: message handling failed")
	ErrObservedOperation      = errors.New("messaging: observed operation failed")
)

Functions

func AsGoError 

func AsGoError(err error) *gerrors.Error

AsGoError projects err into the shared structured error contract without changing the error returned by the messaging API. Existing go-errors values pass through unchanged. Package errors retain errors.Is/errors.As behavior through a source wrapper whose Error string is safe to expose.

func AsRetryableError 

func AsRetryableError(err error) *gerrors.RetryableError

AsRetryableError projects err as retryable only when its stable messaging class proves that retrying is safe. In particular, ambiguous publications are never retryable even when a transport marks the failure temporary. A non-nil return value is guaranteed to report IsRetryable() == true.

func ErrorSlogAttributes 

func ErrorSlogAttributes(err error) []slog.Attr

ErrorSlogAttributes returns structured attributes safe for library logging. It deliberately drops sources, stack/location data, request IDs, validation payloads, and metadata outside the messaging whitelist.

func IsMessageError 

func IsMessageError(err error) bool

IsMessageError reports whether consumption can continue after err.

func NewMessageError 

func NewMessageError(cause error) error

NewMessageError classifies a recoverable error associated with one consumed message without exposing its potentially sensitive cause in Error().

Types

type Acknowledger 

type Acknowledger interface {
	Ack(context.Context) error
	Nack(context.Context, NackOptions) error
}

type BasicDelivery 

type BasicDelivery struct {
	// contains filtered or unexported fields
}

func NewDelivery 

func NewDelivery(envelope Envelope, info DeliveryInfo) BasicDelivery

func (BasicDelivery) Envelope 

func (d BasicDelivery) Envelope() Envelope

func (BasicDelivery) Info 

func (d BasicDelivery) Info() DeliveryInfo

type Capabilities 

type Capabilities struct {
	Durability         bool
	Acknowledgement    bool
	CompetingConsumers bool
	Ordering           bool
	Replay             bool
	RequestReply       bool
	Delay              bool
	Fanout             bool
}

func (Capabilities) Supports 

func (c Capabilities) Supports(capability Capability) bool

func (Capabilities) Validate 

func (c Capabilities) Validate(required ...Capability) error

type Capability 

type Capability string
const (
	CapabilityDurability         Capability = "durability"
	CapabilityAcknowledgement    Capability = "acknowledgement"
	CapabilityCompetingConsumers Capability = "competing_consumers"
	CapabilityOrdering           Capability = "ordering"
	CapabilityReplay             Capability = "replay"
	CapabilityRequestReply       Capability = "request_reply"
	CapabilityDelay              Capability = "delay"
	CapabilityFanout             Capability = "fanout"
)

func CapabilityForDisposition 

func CapabilityForDisposition(disposition Disposition) []Capability

type Codec 

type Codec interface {
	Encode(context.Context, Envelope) ([]byte, error)
	Decode(context.Context, []byte) (Envelope, error)
	ContentType() string
}

Codec encodes and decodes complete transport envelopes.

type ConsumeDriver 

type ConsumeDriver interface {
	Driver
	Consumer
}

type Consumer 

type Consumer interface {
	Subscribe(context.Context, Source, Handler) (Subscription, error)
}

type CorrelationRegistry 

type CorrelationRegistry struct {
	// contains filtered or unexported fields
}

func NewCorrelationRegistry 

func NewCorrelationRegistry(capacity int, defaultTimeout time.Duration, observers ...Observer) (*CorrelationRegistry, error)

func (*CorrelationRegistry) Deliver 

func (r *CorrelationRegistry) Deliver(envelope Envelope) bool

Deliver resolves exactly one matching waiter. False identifies late, duplicate, unknown, or mismatched replies; mismatches leave the real waiter active.

func (*CorrelationRegistry) Pending 

func (r *CorrelationRegistry) Pending() int

func (*CorrelationRegistry) Register 

func (r *CorrelationRegistry) Register(correlationID, expectedType string, deadline time.Time) (*ReplyWaiter, error)

Register must be called before publishing the corresponding request.

type Delivery 

type Delivery interface {
	Envelope() Envelope
	Info() DeliveryInfo
}

Delivery exposes clones so handlers cannot mutate driver-owned values.

type DeliveryInfo 

type DeliveryInfo struct {
	Transport   string
	Destination string
	DeliveryID  string
	Attempt     int
	ReceivedAt  time.Time
	Metadata    map[string]string
}

func (DeliveryInfo) Clone 

func (i DeliveryInfo) Clone() DeliveryInfo

type Destination 

type Destination struct {
	Name string
}

type Disposition 

type Disposition string
const (
	DispositionComplete   Disposition = "complete"
	DispositionRetry      Disposition = "retry"
	DispositionReject     Disposition = "reject"
	DispositionDeadLetter Disposition = "dead_letter"
)

type Driver 

type Driver interface {
	Capabilities() Capabilities
	Start(context.Context) error
	Ready() <-chan struct{}
	Errors() <-chan error
	Close(context.Context) error
}

Driver implementations must close Ready exactly once after their provider is usable. Start, Capabilities and Close must be safe for concurrent callers.

type DriverRegistry 

type DriverRegistry struct {
	// contains filtered or unexported fields
}

DriverRegistry atomically publishes complete immutable driver snapshots.

func NewDriverRegistry 

func NewDriverRegistry(drivers map[string]Driver) (*DriverRegistry, error)

func (*DriverRegistry) All 

func (r *DriverRegistry) All() map[string]Driver

func (*DriverRegistry) Lookup 

func (r *DriverRegistry) Lookup(name string) (Driver, bool)

func (*DriverRegistry) Replace 

func (r *DriverRegistry) Replace(drivers map[string]Driver) error

func (*DriverRegistry) Require 

func (r *DriverRegistry) Require(name string, required ...Capability) (Driver, error)

type Envelope 

type Envelope struct {
	ID            string            `json:"id"`
	Type          string            `json:"type"`
	Kind          Kind              `json:"kind"`
	SchemaVersion string            `json:"schema_version"`
	Payload       []byte            `json:"-"`
	ContentType   string            `json:"content_type"`
	Timestamp     time.Time         `json:"timestamp"`
	CorrelationID string            `json:"correlation_id,omitempty"`
	CausationID   string            `json:"causation_id,omitempty"`
	ReplyTo       string            `json:"reply_to,omitempty"`
	Deadline      time.Time         `json:"deadline"`
	Headers       map[string]string `json:"headers,omitempty"`
}

Envelope is the transport-neutral message exchanged across driver boundaries. Use NewEnvelope or Clone before handing values to independently-owned code.

func NewEnvelope 

func NewEnvelope(id, messageType string, kind Kind, schemaVersion, contentType string, payload []byte, headers map[string]string) Envelope

NewEnvelope constructs an envelope and defensively copies mutable fields.

func (Envelope) Clone 

func (e Envelope) Clone() Envelope

Clone returns a deep copy suitable for crossing an ownership boundary.

func (Envelope) Validate 

func (e Envelope) Validate() error

Validate checks the envelope using conservative default limits.

func (Envelope) ValidateWith 

func (e Envelope) ValidateWith(limits ValidationLimits) error

type HandleResult 

type HandleResult struct {
	Disposition Disposition
	RetryAfter  time.Duration
	Err         error
}

func Complete 

func Complete() HandleResult

func DeadLetter 

func DeadLetter(err error) HandleResult

func InvokeHandler 

func InvokeHandler(ctx context.Context, handler Handler, delivery Delivery) (result HandleResult)

InvokeHandler contains untrusted handler panics at the delivery boundary. The panic value is deliberately excluded from the returned error because it may contain payload or credential data.

func Reject 

func Reject(err error) HandleResult

func Retry 

func Retry(err error, after time.Duration) HandleResult

type Handler 

type Handler func(context.Context, Delivery) HandleResult

type Ingress 

type Ingress struct {
	// contains filtered or unexported fields
}

func NewIngress 

func NewIngress(drivers *DriverRegistry, bindings []IngressBinding, options ...IngressOption) (*Ingress, error)

func (*Ingress) ReplaceBindings 

func (i *Ingress) ReplaceBindings(bindings []IngressBinding) error

func (*Ingress) Subscribe 

func (i *Ingress) Subscribe(ctx context.Context) ([]Subscription, error)

Subscribe starts every binding from one immutable generation.

type IngressBinding 

type IngressBinding struct {
	Name                 string
	LogicalRoute         string
	Driver               string
	Source               Source
	AcceptedKinds        []Kind
	AcceptedTypes        []string
	AcceptedContentTypes []string
	AcceptedSchemas      []string
	Handlers             []Handler
	RequiredCapabilities []Capability
	RequiredDispositions []Disposition
}

type IngressOption 

type IngressOption func(*Ingress)

func WithIngressObserver 

func WithIngressObserver(observer Observer) IngressOption

type JSONCodec 

type JSONCodec struct {
	Limits        ValidationLimits
	MaxFrameBytes int
}

func NewJSONCodec 

func NewJSONCodec() JSONCodec

func (JSONCodec) ContentType 

func (JSONCodec) ContentType() string

func (JSONCodec) Decode 

func (c JSONCodec) Decode(_ context.Context, data []byte) (Envelope, error)

func (JSONCodec) Encode 

func (c JSONCodec) Encode(_ context.Context, envelope Envelope) ([]byte, error)

type Kind 

type Kind string

Kind identifies the delivery intent of an envelope. 

const (
	KindEvent   Kind = "event"
	KindCommand Kind = "command"
	KindQuery   Kind = "query"
	KindReply   Kind = "reply"
)

type LifecycleState 

type LifecycleState string
const (
	LifecycleNew      LifecycleState = "new"
	LifecycleStarting LifecycleState = "starting"
	LifecycleReady    LifecycleState = "ready"
	LifecycleClosing  LifecycleState = "closing"
	LifecycleClosed   LifecycleState = "closed"
	LifecycleFailed   LifecycleState = "failed"
)

type MessageError 

type MessageError struct {
	Cause error
}

MessageError marks a per-message decode, policy, or handler failure. The subscription remains usable; lifecycle and provider errors are deliberately left unwrapped so callers can treat them as terminal.

func (*MessageError) Error 

func (e *MessageError) Error() string

func (*MessageError) Is 

func (e *MessageError) Is(target error) bool

func (*MessageError) Unwrap 

func (e *MessageError) Unwrap() error

type NackOptions 

type NackOptions struct {
	Disposition Disposition
	RetryAfter  time.Duration
	Reason      string
}

type NopObserver 

type NopObserver struct{}

func (NopObserver) Observe 

func (NopObserver) Observe(context.Context, Observation)

type Observation 

type Observation struct {
	Operation     Operation
	LogicalRoute  string
	Kind          Kind
	MessageType   string
	Transport     string
	Destination   string
	CorrelationID string
	Attempt       int
	Outcome       string
	Latency       time.Duration
	Err           error
}

Observation intentionally excludes payload and provider credentials.

type Observer 

type Observer interface {
	Observe(context.Context, Observation)
}

type ObserverFunc 

type ObserverFunc func(context.Context, Observation)

func (ObserverFunc) Observe 

func (f ObserverFunc) Observe(ctx context.Context, observation Observation)

type Operation 

type Operation string
const (
	OperationPublish Operation = "publish"
	OperationConsume Operation = "consume"
	OperationReply   Operation = "reply"
)

type PublishDriver 

type PublishDriver interface {
	Driver
	Publisher
}

type PublishOutcome 

type PublishOutcome string
const (
	PublishAccepted               PublishOutcome = "accepted"
	PublishRejected               PublishOutcome = "rejected"
	PublishDefinitelyNotPublished PublishOutcome = "definitely_not_published"
	PublishAmbiguous              PublishOutcome = "ambiguous"
)

type PublishResult 

type PublishResult struct {
	Outcome           PublishOutcome
	Transport         string
	Destination       string
	ProviderMessageID string
	RecipientCount    *int64
	Metadata          map[string]string
}

func (PublishResult) Clone 

func (r PublishResult) Clone() PublishResult

func (PublishResult) OutcomeError 

func (r PublishResult) OutcomeError() error

OutcomeError converts a non-accepted publish outcome into its stable error classification. Drivers may return both provider-specific errors and an outcome; routers use this method when a driver omits the corresponding error.

type Publisher 

type Publisher interface {
	Publish(context.Context, Destination, Envelope) (PublishResult, error)
}

type ReplyWaiter 

type ReplyWaiter struct {
	// contains filtered or unexported fields
}

func (*ReplyWaiter) Await 

func (w *ReplyWaiter) Await(ctx context.Context) (Envelope, error)

func (*ReplyWaiter) Cancel 

func (w *ReplyWaiter) Cancel()

type Route 

type Route struct {
	Name                   string
	Strategy               Strategy
	Bindings               []RouteBinding
	Kinds                  []Kind
	Types                  []string
	Required               []Capability
	Timeout                time.Duration
	MaxMessageBytes        int
	IdempotencyPolicy      string
	AllowCommandFanout     bool
	AllowAmbiguousFailover bool
}

type RouteBinding 

type RouteBinding struct {
	Driver      string
	Destination Destination
}

type Router 

type Router struct {
	// contains filtered or unexported fields
}

Router resolves logical routes against an immutable configuration generation.

func NewRouter 

func NewRouter(drivers *DriverRegistry, routes []Route, observer Observer) (*Router, error)

func (*Router) Publish 

func (r *Router) Publish(ctx context.Context, logicalRoute string, envelope Envelope) (RoutingResult, error)

func (*Router) ReplaceRoutes 

func (r *Router) ReplaceRoutes(routes []Route) error

func (*Router) Route 

func (r *Router) Route(name string) (Route, bool)

type RoutingResult 

type RoutingResult struct {
	Route    string
	Results  []PublishResult
	Mirrored int
}

type Source 

type Source struct {
	Name     string
	Group    string
	Consumer string
	From     string
}

type Strategy 

type Strategy string
const (
	StrategyPrimary  Strategy = "primary"
	StrategyFailover Strategy = "failover"
	StrategyFanout   Strategy = "fanout"
	StrategyMirror   Strategy = "mirror"
)

type Subscription 

type Subscription interface {
	Ready() <-chan struct{}
	Errors() <-chan error
	Close(context.Context) error
}

type TransportError 

type TransportError struct {
	Class     error
	Transport string
	Operation string
	Temporary bool
	Cause     error
}

func (*TransportError) Error 

func (e *TransportError) Error() string

func (*TransportError) Is 

func (e *TransportError) Is(target error) bool

func (*TransportError) Unwrap 

func (e *TransportError) Unwrap() error

type ValidationLimits 

type ValidationLimits struct {
	MaxPayloadBytes     int
	MaxHeaderCount      int
	MaxHeaderKeyBytes   int
	MaxHeaderValueBytes int
	MaxMetadataBytes    int
	SupportedSchemas    map[string]struct{}
}

ValidationLimits bounds data that is decoded or forwarded by the framework.

func DefaultValidationLimits 

func DefaultValidationLimits() ValidationLimits

Source Files

View all Source files

Directories

Path Synopsis
internal
contracttest
testkit
transport
valkey module

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.