Documentation
¶
Overview ¶
Package redis is a messagequeue publisher and consumer over Redis pub/sub.
This is pub/sub, not Redis Streams, and the difference is the whole of what choosing it commits a caller to.
At-most-once, with no way to notice ¶
Nothing is persisted. A message published while no consumer is subscribed is gone — Redis does not buffer for a subscriber that has not arrived yet, and there is no backlog to read on connect. A handler that returns an error does not see the message again; the error is reported on the consumer's error channel and the loop moves to the next message. There is no acknowledgement, no redelivery, and no dead-letter path, because pub/sub has none to expose.
Work that must not be lost belongs on kafka, pubsub, or sqs, all three of which are at-least-once. What redis is for is fanout where a missed message is tolerable and a broker nobody has to operate is worth more than the guarantee: cache invalidation, presence, live counters.
The one loss this package does close is the subscription race. NewConsumer blocks for a round trip until Redis confirms the SUBSCRIBE has been registered server-side, so a publisher racing a starting consumer cannot slip a message into the window between "we asked to subscribe" and "the server agreed".
Fanout, not work distribution ¶
Every subscriber on a channel receives every message. Two replicas of a service consuming the same topic both run the handler for each message, which is the opposite of what a consumer group does. There is no partitioning and no ordering guarantee beyond what a single Redis connection happens to deliver.
This is why the publisher accepts every messagequeue.PublishOption and honors none. There is nothing here for an ordering key to name — no partitions, no groups, no per-key sequencing — and nothing that deduplicates, so a key given to this publisher changes nothing. It is accepted rather than rejected so that a caller can pass the same options to whichever publisher it was wired with, but a caller who needs ordering needs kafka, pubsub or sqs, not a different call.
Lifecycle ¶
Both providers validate their config at construction and build the client there, so an address list Redis cannot use is a startup error rather than a nil client that panics on first publish. A publisher's Stop is a no-op — the client is shared across every topic, and the provider's Close is what releases it. Consumers close their own subscriptions when their Consume loop exits.
One consumer per topic per provider: a second NewConsumer for the same topic returns messagequeue.ErrConsumerAlreadyRegistered rather than handing back the first caller's consumer wired to the first caller's handler.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Config ¶
type Config struct {
Username string `env:"USERNAME" json:"username,omitempty" yaml:"username,omitempty"`
Password string `env:"PASSWORD" json:"password,omitempty" yaml:"password,omitempty"`
QueueAddresses []string `env:"QUEUE_ADDRESSES" json:"queueAddresses,omitempty" yaml:"queueAddresses,omitempty"`
}
Config configures a Redis-backed consumer.
type ConsumerProvider ¶
type ConsumerProvider struct {
// contains filtered or unexported fields
}
ConsumerProvider is the Redis messagequeue.ConsumerProvider implementation. It is exported, and returned by NewRedisConsumerProvider, so a caller who has chosen Redis can depend on that choice rather than on the interface every broker shares.
func NewRedisConsumerProvider ¶
func NewRedisConsumerProvider(ctx context.Context, cfg Config, opts ...Option) (*ConsumerProvider, error)
NewRedisConsumerProvider returns a ConsumerProvider for a given address.
It takes a context and reports an error so that the config's own ValidateWithContext runs here, and so that an address list redisclient refuses is a startup error. Without either, a config naming no QueueAddresses built cleanly and the provider came back holding a nil client — which nothing noticed until the first Ping panicked.
func (*ConsumerProvider) Close ¶
func (p *ConsumerProvider) Close()
Close closes the shared Redis client, mirroring the publisher provider. Cached consumers close their own subscriptions when their Consume loops exit.
func (*ConsumerProvider) NewConsumer ¶
func (p *ConsumerProvider) NewConsumer(ctx context.Context, topic string, handlerFunc messagequeue.ConsumerFunc) (messagequeue.Consumer, error)
NewConsumer returns a Consumer for a given topic.
type Option ¶
type Option func(*options)
Option configures the providers this package constructs. The zero configuration works: absent observability deps are normalized downstream.
func WithMetricsProvider ¶
WithMetricsProvider attaches a metrics provider.
func WithTracerProvider ¶
WithTracerProvider attaches a tracer provider.
type PublisherProvider ¶
type PublisherProvider struct {
// contains filtered or unexported fields
}
PublisherProvider is the Redis messagequeue.PublisherProvider implementation. It is exported, and returned by NewRedisPublisherProvider, so a caller who has chosen Redis can depend on that choice rather than on the interface every broker shares.
func NewRedisPublisherProvider ¶
func NewRedisPublisherProvider(ctx context.Context, cfg Config, opts ...Option) (*PublisherProvider, error)
NewRedisPublisherProvider returns a PublisherProvider for a given address.
It takes a context and reports an error so that the config's own ValidateWithContext runs here, and so that an address list redisclient refuses is a startup error. Without either, a config naming no QueueAddresses built cleanly and the provider came back holding a nil client — which nothing noticed until the first Publish panicked, in whichever goroutine happened to make it.
func (*PublisherProvider) NewPublisher ¶
func (p *PublisherProvider) NewPublisher(ctx context.Context, topic string) (messagequeue.Publisher, error)
NewPublisher returns a Publisher for a given topic.
Source Files