smsg

package module
v0.1.0 Latest Latest
Warning

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

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

README

smsg

Typed messaging bus for Go, inspired by MassTransit. Part of the s* family: sconf (configuration), srog (logging), shost (hosting), sorm (ORM).

smsg removes the boilerplate around broker clients: consumers are plain generic interfaces, messages travel in envelopes with identifiers and headers, failures are retried with backoff and dead-lettered, and the broker sits behind a small transport SPI — swap RabbitMQ for Kafka (or an in-memory transport in tests) without touching consumer code.

go get github.com/dvislobokov/smsg          # core + inmem + smsgtest (stdlib only)
go get github.com/dvislobokov/smsg/rabbit   # RabbitMQ transport (amqp091-go)
go get github.com/dvislobokov/smsg/kafka    # Kafka transport (franz-go)

Quick start

package main

import (
	"context"
	"os"

	"github.com/dvislobokov/shost"
	"github.com/dvislobokov/smsg"
	"github.com/dvislobokov/smsg/rabbit"
	"github.com/dvislobokov/srog"
)

type OrderCreated struct {
	ID    string
	Total float64
}

type BillingConsumer struct{ /* dependencies */ }

func (c *BillingConsumer) Consume(ctx context.Context, m smsg.Message[OrderCreated]) error {
	// m.Body is the typed payload; m.ID, m.CorrelationID, m.Headers,
	// m.Attempt carry the envelope metadata.
	return charge(ctx, m.Body.ID, m.Body.Total)
}

func main() {
	log := srog.MustNew(srog.WithConsole()) // satisfies smsg.Logger directly

	bus := smsg.New().
		WithName("orders-bus").
		WithLogger(log).
		WithTransport(rabbit.New(rabbit.Config{URL: os.Getenv("AMQP_URL")})).
		AddConsumer(smsg.For[OrderCreated](&BillingConsumer{}),
			smsg.Topic("orders"), smsg.Group("billing"),
			smsg.Concurrency(8),
			smsg.Retry(smsg.RetryPolicy{MaxAttempts: 5}),
			smsg.DeadLetter("orders.dlq")).
		MustBuild()

	// The bus is a shost service: graceful shutdown drains in-flight
	// messages; restart supervision re-dials a lost broker.
	shost.New().
		WithLogger(log).
		AddService(bus, shost.WithRestart(shost.RestartPolicy{})).
		MustBuild().
		Run()
}

Publishing, from anywhere the bus is available:

err := bus.Publish(ctx, OrderCreated{ID: "42", Total: 9.99},
	smsg.WithCorrelationID(requestID))

The topic defaults to the message type name (OrderCreated); override it with smsg.ToTopic("orders") or a bus-wide WithTopicNamer.

Testing without a broker

The inmem transport and the smsgtest helpers run the bus fully in-process with the same semantics — fan-out across groups, competing consumers within one, retry, dead-lettering:

func TestBilling(t *testing.T) {
	col := smsgtest.NewCollector[OrderCreated]()
	bus := smsgtest.Start(t, smsg.New().
		WithTransport(inmem.New()).
		AddConsumer(col.Registration(), smsg.Topic("orders")))

	bus.Publish(OrderCreated{ID: "42", Total: 9.99}, smsg.ToTopic("orders"))

	got := col.Wait(t, 1)
	if got[0].Body.ID != "42" {
		t.Fatalf("wrong order: %+v", got[0])
	}
}

smsgtest.NewRecorder() additionally records bus events (published, consumed, retry-scheduled, dead-lettered, …) for assertions.

Concepts

smsg RabbitMQ Kafka
Topic durable exchange (fanout by default) topic
Group durable queue <topic>.<group> bound to the exchange consumer group
Fan-out every group's queue gets a copy every group reads the full log
Competing consumers one queue, prefetch = Concurrency partitions within the group
Ack Ack after the bus accepts the message explicit offset commit
  • Envelope — every message carries a MessageID, optional CorrelationID, the MessageType (used to skip foreign types on shared topics), ContentType, timestamp and user headers. Serialization is JSON by default (WithSerializer to replace).
  • Retry — in-process with exponential backoff, identical on every transport: smsg.Retry(smsg.RetryPolicy{MaxAttempts: 5}) per consumer or WithRetry bus-wide.
  • Dead-lettering — after the policy is exhausted the envelope is published to the DeadLetter topic with smsg-error, smsg-attempts and smsg-origin-topic headers, then acknowledged. Without a dead-letter topic the message is dropped with a warning (never a silent poison loop); OnExhausted(smsg.ExhaustedRequeue) opts into broker redelivery instead.
  • Transport options — the core stays broker-agnostic; transports add typed sugar: kafka.PartitionKey("42"), kafka.StartOffsetFirst(), rabbit.ExchangeType("topic"), rabbit.RoutingKey("orders.eu.berlin"), rabbit.BindingKey("orders.eu.#").
  • Observability — WithObserver(smsg.Observer{...}) receives every lifecycle and message event for metrics/tracing with zero telemetry dependencies in the core; WithLogger accepts *srog.Logger directly or smsg.SlogLogger(slog.Default()).

Integration tests

Core and inmem tests need nothing. The transport tests skip unless a broker is reachable:

docker compose up -d
SMSG_RABBIT_URL=amqp://guest:guest@localhost:5672/ go test ./rabbit/...
SMSG_KAFKA_BROKERS=localhost:9092 go test ./kafka/...

Modules

The core module (with inmem and smsgtest) depends only on the standard library. Broker clients live in submodules with their own version streams:

Module Dependency
github.com/dvislobokov/smsg —
github.com/dvislobokov/smsg/rabbit rabbitmq/amqp091-go
github.com/dvislobokov/smsg/kafka twmb/franz-go

See examples/ for runnable inmem, RabbitMQ and Kafka programs.

License

MIT

Documentation

Overview

Package smsg is a typed messaging bus for Go, modeled after MassTransit: consumers are plain generic interfaces, messages travel in envelopes with identifiers and headers, failed handling is retried in-process and dead-lettered, and the broker is hidden behind a small transport SPI. The inmem subpackage runs the bus fully in-process for tests; the rabbit and kafka submodules connect it to real brokers.

Index

Constants

View Source 
const (
	DefaultRetryInitialDelay = 100 * time.Millisecond
	DefaultRetryMaxDelay     = 10 * time.Second
	DefaultRetryFactor       = 2.0
)

Defaults applied to zero-valued RetryPolicy fields.

View Source 
const (
	HeaderError       = "smsg-error"
	HeaderAttempts    = "smsg-attempts"
	HeaderOriginTopic = "smsg-origin-topic"
)

Headers stamped onto dead-lettered envelopes.

View Source 
const DefaultName = "smsg"

DefaultName is the bus name when Builder.WithName was not called.

Variables

View Source 
var (
	// ErrNotStarted is returned by Publish before Start has been called.
	ErrNotStarted = errors.New("smsg: bus not started")
	// ErrStopped is returned by Publish after the bus has been stopped,
	// and by Start when the bus has already run once.
	ErrStopped = errors.New("smsg: bus stopped")
	// ErrNoTopic is returned by Publish when no topic could be resolved:
	// the message type has no name (anonymous type) and no ToTopic option
	// was given.
	ErrNoTopic = errors.New("smsg: no topic resolved for message")
)

Sentinel errors returned by Bus operations. Test with errors.Is.

Functions

func DefaultTopicNamer 

func DefaultTopicNamer(t reflect.Type) string

DefaultTopicNamer names the topic after the plain Go type name: OrderCreated -> "OrderCreated". Anonymous types yield "".

Types

type Builder 

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

Builder assembles a Bus. All methods return the receiver for chaining; configuration errors are accumulated and reported by Build.

func New 

func New() *Builder

New creates an empty Builder.

func (*Builder) AddConsumer 

func (b *Builder) AddConsumer(c ConsumerRegistration, opts ...ConsumerOption) *Builder

AddConsumer registers an erased consumer (see For and Handle) with its subscription options. Each registration becomes one subscription; the (topic, group) pair must be unique per bus.

func (*Builder) Build 

func (b *Builder) Build() (*Bus, error)

Build validates the configuration and returns the Bus.

func (*Builder) MustBuild 

func (b *Builder) MustBuild() *Bus

MustBuild is Build panicking on configuration errors. Intended for main().

func (*Builder) WithLogger 

func (b *Builder) WithLogger(l Logger) *Builder

WithLogger sets the logger for bus lifecycle and message-flow events. Without it the bus stays silent; errors are still returned and observable.

func (*Builder) WithName 

func (b *Builder) WithName(name string) *Builder

WithName sets the bus name reported by Bus.Name — the service name when the bus is hosted by shost. Defaults to DefaultName.

func (*Builder) WithObserver 

func (b *Builder) WithObserver(o Observer) *Builder

WithObserver registers a lifecycle observer (see Observer). Multiple observers are invoked in registration order.

func (*Builder) WithRetry 

func (b *Builder) WithRetry(p RetryPolicy) *Builder

WithRetry sets the bus-wide default retry policy, overridable per consumer with the Retry option. Without it messages are not retried.

func (*Builder) WithSerializer 

func (b *Builder) WithSerializer(s Serializer) *Builder

WithSerializer sets the body serializer. Defaults to JSON().

func (*Builder) WithTopicNamer 

func (b *Builder) WithTopicNamer(fn TopicNamer) *Builder

WithTopicNamer sets how topics are derived from message types. Defaults to DefaultTopicNamer (the plain type name).

func (*Builder) WithTransport 

func (b *Builder) WithTransport(t Transport) *Builder

WithTransport sets the transport the bus runs on. Required; exactly one.

type Bus 

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

Bus publishes messages and runs consumer subscriptions on one transport. Its lifecycle satisfies the shost.Service contract structurally (Name/Start/Stop plus Ready), so a Bus can be passed to shost.Builder.AddService without smsg importing shost.

A Bus runs once: Start after Stop returns ErrStopped.

func (*Bus) Name 

func (b *Bus) Name() string

Name reports the bus name (shost.Service contract).

func (*Bus) Publish 

func (b *Bus) Publish(ctx context.Context, msg any, opts ...PublishOption) error

Publish serializes the message and sends it to its topic. The topic is derived from the message type via the bus TopicNamer unless overridden with ToTopic. Publish is safe for concurrent use once Start has run; before Start it returns ErrNotStarted, after Stop it returns ErrStopped.

func (*Bus) Ready 

func (b *Bus) Ready() <-chan struct{}

Ready is closed once the transport is connected and every subscription is running (shost.Readier contract).

func (*Bus) Start 

func (b *Bus) Start(ctx context.Context) error

Start connects the transport, starts all consumer subscriptions and blocks for the bus lifetime (shost.Service contract). It returns nil after ctx is canceled or Stop is called, and an error when the transport fails to connect, a subscription cannot start, or a running subscription dies (lost broker connection) — letting a supervising host restart it.

func (*Bus) Stop 

func (b *Bus) Stop(ctx context.Context) error

Stop closes all subscriptions in reverse order — stopping fetching and draining in-flight messages within the ctx deadline — then closes the transport (shost.Service contract). Safe to call more than once.

type Consumer 

type Consumer[T any] interface {
	Consume(ctx context.Context, m Message[T]) error
}

Consumer handles messages of one type, mirroring MassTransit's IConsumer<T>. Implementations must be safe for concurrent use when registered with Concurrency > 1.

type ConsumerFunc 

type ConsumerFunc[T any] func(ctx context.Context, m Message[T]) error

ConsumerFunc adapts a plain function to the Consumer interface.

func (ConsumerFunc[T]) Consume 

func (f ConsumerFunc[T]) Consume(ctx context.Context, m Message[T]) error

Consume implements Consumer.

type ConsumerOption 

type ConsumerOption func(*consumerConfig)

ConsumerOption customizes a single consumer registration.

func Concurrency 

func Concurrency(n int) ConsumerOption

Concurrency sets the maximum number of concurrent handler invocations for this consumer. Defaults to 1.

func ConsumerMeta 

func ConsumerMeta(key, value string) ConsumerOption

ConsumerMeta attaches a namespaced transport hint to the subscription ("rabbit.exchange", "kafka.start-offset"). Transport submodules export typed sugar over this.

func DeadLetter 

func DeadLetter(topic string) ConsumerOption

DeadLetter names the topic exhausted messages are published to. Setting it makes ExhaustedDeadLetter the default exhausted action.

func Group 

func Group(name string) ConsumerOption

Group names the competing-consumer group (Kafka consumer group, RabbitMQ queue). Defaults to the topic name: one group per topic.

func OnExhausted 

func OnExhausted(a ExhaustedAction) ConsumerOption

OnExhausted overrides what happens once the retry policy is exhausted. Defaults: ExhaustedDeadLetter when DeadLetter is set, ExhaustedDrop otherwise.

func Retry 

func Retry(p RetryPolicy) ConsumerOption

Retry sets the in-process retry policy for this consumer, overriding the bus-wide policy from Builder.WithRetry.

func Topic 

func Topic(name string) ConsumerOption

Topic overrides the topic the consumer subscribes to. Without it the topic is derived from the message type via the bus TopicNamer.

type ConsumerRegistration 

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

ConsumerRegistration is an erased, transport-ready form of a typed consumer. Values are produced by For or Handle and passed to Builder.AddConsumer; the zero value is invalid.

func For 

func For[T any](c Consumer[T]) ConsumerRegistration

For erases a typed consumer into a registration the bus can run. The message type's name becomes the default topic (via the bus TopicNamer) and the envelope MessageType to match on shared topics.

func Handle 

func Handle[T any](fn func(ctx context.Context, m Message[T]) error) ConsumerRegistration

Handle is For with a plain function: Handle[T](fn) is For[T](ConsumerFunc[T](fn)).

type Delivery 

type Delivery struct {
	Envelope *Envelope
	// Attempt is the transport-level redelivery count when the broker
	// reports one (RabbitMQ x-death / Redelivered), 0 when unknown.
	Attempt int
}

Delivery is one received message handed from a transport to the bus.

type DeliveryHandler 

type DeliveryHandler func(ctx context.Context, d *Delivery) error

DeliveryHandler processes one delivery on behalf of the bus. A nil return means success: the transport must acknowledge the message (ack / offset commit). A non-nil return means the bus could neither handle nor dead-letter the message; the transport applies its native redelivery mechanism. Transports may invoke the handler concurrently, up to SubscriptionSpec.Concurrency calls at a time.

type Envelope 

type Envelope struct {
	// MessageID uniquely identifies the message; generated at publish
	// when empty.
	MessageID string
	// CorrelationID links the message to the request or saga that
	// produced it. Optional.
	CorrelationID string
	// MessageType is the logical type name (usually the Go type name).
	// Consumers skip deliveries whose MessageType does not match theirs,
	// which allows several message types to share one topic.
	MessageType string
	// ContentType is the MIME type of Body, e.g. "application/json".
	ContentType string
	// Topic the message was published to.
	Topic string
	// Timestamp is the publish time in UTC.
	Timestamp time.Time
	// Headers are user-defined key/value pairs carried on the wire.
	Headers map[string]string
	// Metadata carries transport hints ("kafka.partition-key",
	// "rabbit.routing-key"). It is NOT serialized onto the wire;
	// transports read the keys in their namespace and ignore the rest.
	Metadata map[string]string
	// Body is the serialized message payload.
	Body []byte
}

Envelope is the transport-level representation of a message: the serialized body plus the metadata that travels with it. Transports map its fields onto their native wire format (AMQP properties, Kafka headers) and reconstruct it on the consuming side.

func (*Envelope) Clone 

func (e *Envelope) Clone() *Envelope

Clone returns a deep copy of the envelope. Transports fanning one envelope out to several consumers copy it first so handlers cannot observe each other's mutations.

type ExhaustedAction 

type ExhaustedAction int

ExhaustedAction is what the bus does with a message once its retry policy is exhausted. 

const (
	// ExhaustedDeadLetter publishes the message to the consumer's
	// dead-letter topic and acknowledges the original. It is the default
	// when the DeadLetter option is set.
	ExhaustedDeadLetter ExhaustedAction = iota
	// ExhaustedRequeue hands the error back to the transport, which
	// applies its native redelivery (RabbitMQ nack+requeue, Kafka
	// uncommitted offset). Opt-in: it can loop a poison message forever.
	ExhaustedRequeue
	// ExhaustedDrop acknowledges the message and logs a warning. It is
	// the default when no dead-letter topic is configured, so a poison
	// message never blocks the subscription silently.
	ExhaustedDrop
)

type Logger 

type Logger interface {
	Debug(template string, args ...any)
	Information(template string, args ...any)
	Warning(template string, args ...any)
	Error(err error, template string, args ...any)
}

Logger is the minimal logging surface the bus needs for lifecycle and message-flow events. The method set is signature-compatible with srog, so an *srog.Logger can be passed to Builder.WithLogger directly; any other logging library can be adapted with a small wrapper.

Templates use srog-style named placeholders: "consumed {Topic}".

func SlogLogger 

func SlogLogger(l *slog.Logger) Logger

SlogLogger adapts a *slog.Logger to the smsg Logger interface for applications not using srog. srog-style message templates are rendered into the log message, and each named placeholder additionally becomes a slog attribute: 

// "consumed orders" with attribute Topic=orders
log.Debug("consumed {Topic}", "orders")

type Message 

type Message[T any] struct {
	// Body is the deserialized payload.
	Body T
	// ID is the unique message identifier from the envelope.
	ID string
	// CorrelationID links the message to the request or saga that
	// produced it; empty when the publisher did not set one.
	CorrelationID string
	// Topic the message arrived on.
	Topic string
	// Timestamp is the publish time in UTC.
	Timestamp time.Time
	// Headers are the user-defined key/value pairs from the envelope.
	Headers map[string]string
	// Attempt is the 1-based attempt number across in-process retries.
	Attempt int
}

Message is one deserialized delivery as seen by a typed consumer.

type Observer 

type Observer struct {
	BusStarted          func()
	BusStopped          func(err error)
	SubscriptionStarted func(topic, group string)
	SubscriptionStopped func(topic, group string, err error)
	Published           func(topic string, env *Envelope, elapsed time.Duration, err error)
	Consumed            func(topic, group, messageID string, attempt int, elapsed time.Duration, err error)
	RetryScheduled      func(topic, group, messageID string, attempt int, delay time.Duration, err error)
	DeadLettered        func(topic, group, dlqTopic, messageID string, err error)
	Dropped             func(topic, group, messageID string, err error)
}

Observer receives bus lifecycle and message-flow events for metrics and tracing without coupling the core to any telemetry library. All fields are optional; nil callbacks are skipped. Callbacks run synchronously on the bus hot path and must be fast; panics are recovered and logged.

type PublishOption 

type PublishOption func(*publishConfig)

PublishOption customizes a single Publish call.

func Meta 

func Meta(key, value string) PublishOption

Meta attaches a namespaced transport hint to the envelope ("kafka.partition-key", "rabbit.routing-key"). Not serialized onto the wire; transport submodules export typed sugar over this.

func ToTopic 

func ToTopic(topic string) PublishOption

ToTopic overrides the topic the message is published to. Without it the topic is derived from the message type via the bus TopicNamer.

func WithCorrelationID 

func WithCorrelationID(id string) PublishOption

WithCorrelationID stamps the correlation identifier onto the envelope.

func WithHeader 

func WithHeader(key, value string) PublishOption

WithHeader adds a user header carried on the wire.

func WithMessageID 

func WithMessageID(id string) PublishOption

WithMessageID overrides the generated message identifier.

type RetryPolicy 

type RetryPolicy struct {
	// MaxAttempts is the total number of attempts including the first.
	// 0 means 1: no retry.
	MaxAttempts int
	// InitialDelay is the pause before the first retry.
	InitialDelay time.Duration
	// MaxDelay caps the exponential backoff.
	MaxDelay time.Duration
	// Factor multiplies the delay after each consecutive retry.
	Factor float64
}

RetryPolicy controls in-process redelivery of a message whose consumer returned an error. Retries happen inside the consuming process with exponential backoff, identically on every transport. Zero-valued fields take the Default* constants above.

type Serializer 

type Serializer interface {
	// ContentType is the MIME type stamped into Envelope.ContentType.
	ContentType() string
	Serialize(v any) ([]byte, error)
	Deserialize(data []byte, v any) error
}

Serializer converts message bodies to and from their wire representation. The bus uses one serializer for both publishing and consuming; the default is JSON.

func JSON 

func JSON() Serializer

JSON returns the default serializer backed by encoding/json.

type Subscription 

type Subscription interface {
	// Close stops fetching new messages and waits for in-flight handlers
	// to finish, bounded by ctx.
	Close(ctx context.Context) error
	// Done is closed when the subscription ends for any reason.
	Done() <-chan struct{}
	// Err reports why the subscription ended: nil after Close, non-nil
	// after an abnormal end (lost connection, fatal fetch error).
	Err() error
}

Subscription is one active consumer subscription.

type SubscriptionSpec 

type SubscriptionSpec struct {
	Topic string
	// Group names the competing-consumer group: the Kafka consumer group,
	// or the RabbitMQ queue bound to the topic exchange. Messages are
	// load-balanced within a group and fanned out across groups.
	Group string
	// Concurrency is the maximum number of concurrent handler
	// invocations; >= 1 (the builder normalizes it).
	Concurrency int
	// Metadata carries namespaced transport hints ("rabbit.exchange",
	// "kafka.start-offset"). Transports ignore keys outside their
	// namespace.
	Metadata map[string]string
}

SubscriptionSpec describes one consumer subscription for a transport.

type TopicNamer 

type TopicNamer func(t reflect.Type) string

TopicNamer derives a topic name from a message type. It is used when neither the Topic consumer option nor the ToTopic publish option names the topic explicitly. Returning "" means the topic cannot be derived (Publish then fails with ErrNoTopic; AddConsumer fails at Build).

type Transport 

type Transport interface {
	// Name identifies the transport in logs, e.g. "rabbit".
	Name() string
	// Connect establishes the broker connection. Called once by
	// Bus.Start before any Subscribe or Publish.
	Connect(ctx context.Context) error
	// Publish sends the envelope to the topic, creating broker topology
	// on first use where the broker requires it.
	Publish(ctx context.Context, topic string, env *Envelope) error
	// Subscribe starts delivering messages matching spec to h.
	Subscribe(ctx context.Context, spec SubscriptionSpec, h DeliveryHandler) (Subscription, error)
	// Close releases the broker connection. Called by Bus.Stop after all
	// subscriptions are closed.
	Close(ctx context.Context) error
}

Transport moves opaque envelopes to and from a broker. Implementations: the inmem subpackage (in-process, for tests), and the rabbit and kafka submodules. All methods must be safe for concurrent use.

Source Files

View all Source files

Directories

Path Synopsis
Package inmem is an in-process smsg transport for tests and examples: no broker, no network, no extra dependencies.
 Package inmem is an in-process smsg transport for tests and examples: no broker, no network, no extra dependencies.
kafka module
rabbit module
Package smsgtest provides test helpers for smsg applications: running a bus inside a test with automatic cleanup, recording bus events for assertions, and collecting consumed messages.
 Package smsgtest provides test helpers for smsg applications: running a bus inside a test with automatic cleanup, recording bus events for assertions, and collecting consumed messages.

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.