Documentation
¶
Overview ¶
Package messaging provides a factory for message publisher and subscriber implementations with multiple backends.
Use NewFromConfig() for app wiring (the tier-root factory that builds a publisher straight from the loaded config), or the lower-level NewPublisher(), NewSubscriber(), and NewAdapter() constructors. The factory selects the implementation at runtime based on Kind; backend impls live in the sqs/ subpackage.
Example:
pub, err := messaging.NewPublisher(messaging.KindSQS, messaging.WithQueueURL(url))
if err != nil {
log.Fatal(err)
}
Index ¶
- func NewAdapter(kind Kind, opts ...options.Option[Config]) (*sqs.Adapter, error)
- func NewFromConfig(ctx context.Context, kind Kind, cfg infra.SQSConfig, ...) (interfaces.MessagePublisher, error)
- func NewPublisher(kind Kind, opts ...options.Option[Config]) (interfaces.MessagePublisher, error)
- func NewSubscriber(kind Kind, opts ...options.Option[Config]) (interfaces.MessageConsumer, error)
- func WithAWSRegion(region string) options.Option[Config]
- func WithEndpoint(endpoint string) options.Option[Config]
- func WithLogger(logger interfaces.Logger) options.Option[Config]
- func WithMetrics(metrics interfaces.Metrics) options.Option[Config]
- func WithQueueURL(url string) options.Option[Config]
- func WithRedisClient(client *goredis.Client) options.Option[Config]
- func WithRetrier(retrier interfaces.Retrier) options.Option[Config]
- func WithStaticCredentials(accessKeyID, secretAccessKey string) options.Option[Config]
- type Config
- type Kind
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func NewAdapter ¶
NewAdapter creates a messaging adapter of the specified kind with optional functional options. Returns an error if the kind is unknown.
func NewFromConfig ¶
func NewFromConfig( ctx context.Context, kind Kind, cfg infra.SQSConfig, opts ...options.Option[Config], ) (interfaces.MessagePublisher, error)
NewFromConfig builds a MessagePublisher selected by the tier kind — the SQS backend (KindSQS, configured from cfg), the in-process stub (KindMemory, ignoring cfg), or Redis Pub/Sub (KindRedis, whose shared go-redis client is injected via the WithRedisClient option — NOT from cfg; the SQSConfig arg is inert for Redis) — wrapping the thin core in the messaging decorators. It is the app-wiring entrypoint, sitting at the tier root above the backends exactly like cache/storage NewFromConfig; additional options (logger, metrics, retrier, redis client) may be layered on. The context bounds the SQS backend's AWS config load; the other backends do no I/O at construction.
func NewPublisher ¶
func NewPublisher( kind Kind, opts ...options.Option[Config], ) (interfaces.MessagePublisher, error)
NewPublisher creates a message publisher of the specified kind with optional functional options. Returns an error if the kind is unknown.
func NewSubscriber ¶
func NewSubscriber( kind Kind, opts ...options.Option[Config], ) (interfaces.MessageConsumer, error)
NewSubscriber creates a message subscriber of the specified kind with optional functional options. Returns an error if the kind is unknown.
func WithAWSRegion ¶
WithAWSRegion sets the AWS region for the SQS backend.
func WithEndpoint ¶
WithEndpoint overrides the SQS service endpoint (e.g. ElasticMQ at "http://localhost:9324"). Empty uses the default AWS endpoint for the region.
func WithLogger ¶
func WithLogger(logger interfaces.Logger) options.Option[Config]
WithLogger sets the logger that decorates the publisher with failure logging and is passed to the SQS/Redis subscribers to report receive/handle errors (their goroutines have nowhere else to surface them).
func WithMetrics ¶
func WithMetrics(metrics interfaces.Metrics) options.Option[Config]
WithMetrics sets the metrics registry that decorates the publisher with publish counters + latency and the SQS/Redis subscribers with per-message handle metrics.
func WithQueueURL ¶
WithQueueURL sets the base queue URL for the SQS backend.
func WithRedisClient ¶
WithRedisClient injects the shared go-redis client used by the Redis Pub/Sub backend (KindRedis), typically the cache tier's Cache.Client(). The backend never dials its own connection, so this is required when Kind is KindRedis.
func WithRetrier ¶
func WithRetrier(retrier interfaces.Retrier) options.Option[Config]
WithRetrier sets the retrier that decorates the publisher, retrying a failed publish per the retrier's policy.
Types ¶
type Config ¶
type Config struct {
// Redis configures the Redis Pub/Sub backend (used when Kind is KindRedis).
Redis redis.Config
// Logger decorates the publisher with failure logging and is passed to the
// subscribers to surface receive/handle errors.
Logger interfaces.Logger
// Metrics decorates the publisher and subscribers with counters + latency.
Metrics interfaces.Metrics
// Retrier decorates the publisher, retrying a failed publish per its policy.
Retrier interfaces.Retrier
// SQS configures the SQS backend (used when Kind is KindSQS).
SQS sqs.Config
// Kind selects which backend the factory builds (KindSQS or KindRedis).
Kind Kind
}
Config is the top-level configuration for the messaging package. (Field order is fieldalignment-optimized.)
func DefaultConfig ¶
func DefaultConfig() Config
DefaultConfig returns the default messaging configuration with KindSQS and us-east-1 region.
type Kind ¶
type Kind int
Kind specifies which messaging implementation to use.
const ( // KindSQS uses AWS SQS for message publishing and subscription. Suitable for // distributed deployments with AWS infrastructure. KindSQS Kind = iota // KindMemory uses an in-process broker (dev/test/all-stubs; no external broker). KindMemory // KindRedis uses Redis Pub/Sub (fire-and-forget broadcast + dynamic SUBSCRIBE/ // PSUBSCRIBE). Its go-redis client is injected via WithRedisClient (shared with // the cache tier), not built from config — the backend for cross-replica fan-out // (e.g. pushing a message to whichever replica holds a client connection). KindRedis )
func ParseKind ¶
ParseKind maps a config kind string ("sqs" | "memory" | "redis") to its Kind, failing loudly on an unknown value. It is the messaging-tier bridge a composition root uses to select a backend from configuration (ARCHITECTURE.md#swappable-components); the backend config (e.g. the injected Redis client) is still supplied separately via options.
Source Files
Directories
¶
| Path | Synopsis |
|---|---|
|
Package decorators wraps the messaging MessagePublisher and MessageHandler with cross-cutting concerns.
|
Package decorators wraps the messaging MessagePublisher and MessageHandler with cross-cutting concerns. |
|
Package memory provides an in-process messaging backend for no-infra builds and tests.
|
Package memory provides an in-process messaging backend for no-infra builds and tests. |
|
Package redis provides a Redis Pub/Sub messaging integration (publisher + subscriber) implementing the core MessagePublisher / MessageConsumer contracts over PUBLISH / SUBSCRIBE / PSUBSCRIBE.
|
Package redis provides a Redis Pub/Sub messaging integration (publisher + subscriber) implementing the core MessagePublisher / MessageConsumer contracts over PUBLISH / SUBSCRIBE / PSUBSCRIBE. |
|
Package sqs provides an AWS-SDK-backed SQS messaging integration (publisher + subscriber) usable against real AWS SQS and against ElasticMQ for local dev.
|
Package sqs provides an AWS-SDK-backed SQS messaging integration (publisher + subscriber) usable against real AWS SQS and against ElasticMQ for local dev. |