redisstream

Documentation

Overview

Package redisstream provides a Redis Streams backend for the ling-base/mq broker abstraction. It supports consumer groups (XREADGROUP), standalone consumption (XREAD), stream trimming (XTRIM/MAXLEN), and manual acknowledgement (XACK).

Redis Streams are append-only logs identified by a stream key. In this backend each "exchange" and "queue" maps to a Redis stream key. Streams are auto-created on the first XADD, so DeclareExchange/DeclareQueue are no-ops.

Basic usage

broker, _ := redisstream.New(redisstream.DefaultConfig())
defer broker.Close()
_ = broker.Connect()

producer, _ := broker.Producer("events", mq.PublishOptions{})
_ = producer.Publish(ctx, &mq.Message{Body: []byte(`{"event":"user.created"}`)})

consumer, _ := broker.Consumer("events", mq.ConsumeOptions{
    Handler: func(ctx context.Context, d mq.Delivery) error {
        fmt.Println(string(d.Body()))
        return d.Ack()
    },
})
_ = consumer.Start(ctx)

Index

• type Broker
  • func New(cfg Config) (*Broker, error)
  • func (b *Broker) Bind(queue, exchange, routingKey string) error
  • func (b *Broker) Close() error
  • func (b *Broker) Connect() error
  • func (b *Broker) Consumer(stream string, opts mq.ConsumeOptions) (mq.Consumer, error)
  • func (b *Broker) DeclareExchange(name string, opts mq.ExchangeOptions) error
  • func (b *Broker) DeclareQueue(name string, opts mq.QueueOptions) error
  • func (b *Broker) DeleteExchange(name string) error
  • func (b *Broker) DeleteQueue(name string) error
  • func (b *Broker) IsConnected() bool
  • func (b *Broker) Metrics() mq.Metrics
  • func (b *Broker) Producer(stream string, opts mq.PublishOptions) (mq.Producer, error)
  • func (b *Broker) Unbind(queue, exchange, routingKey string) error
• type Config
  • func DefaultConfig() Config
• type Consumer
  • func (c *Consumer) IsRunning() bool
  • func (c *Consumer) Start(ctx context.Context) error
  • func (c *Consumer) Stop(timeout time.Duration) error
• type Producer
  • func (p *Producer) Close() error
  • func (p *Producer) Publish(ctx context.Context, msg *mq.Message) error

Types

type Broker

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

Broker implements mq.Broker for Redis Streams.

func New

func New(cfg Config) (*Broker, error)

New creates a new Redis Streams broker. Call Connect() to establish the connection.

func (*Broker) Bind

func (b *Broker) Bind(queue, exchange, routingKey string) error

Bind is a no-op. Redis Streams do not have bindings.

func (*Broker) Close

func (b *Broker) Close() error

Close shuts down the broker, closing all producers and consumers and releasing the Redis connection. Close is idempotent.

func (*Broker) Connect

func (b *Broker) Connect() error

Connect establishes the connection to Redis.

func (*Broker) Consumer

func (b *Broker) Consumer(stream string, opts mq.ConsumeOptions) (mq.Consumer, error)

Consumer creates a consumer for the given stream. The consumer is not started; call Start to begin consuming.

If the broker's Group config is non-empty, the consumer operates in consumer-group mode (XREADGROUP). Otherwise it operates in standalone mode (XREAD). The group and consumer name can be overridden via ConsumeOptions.Args with keys "group" and "consumer" (string values).

func (*Broker) DeclareExchange

func (b *Broker) DeclareExchange(name string, opts mq.ExchangeOptions) error

DeclareExchange is a no-op. Redis Streams are auto-created on the first XADD.

func (*Broker) DeclareQueue

func (b *Broker) DeclareQueue(name string, opts mq.QueueOptions) error

DeclareQueue is a no-op. Redis Streams are auto-created on the first XADD.

func (*Broker) DeleteExchange

func (b *Broker) DeleteExchange(name string) error

DeleteExchange is a no-op. Redis Streams do not have exchanges.

func (*Broker) DeleteQueue

func (b *Broker) DeleteQueue(name string) error

DeleteQueue removes a stream by deleting the key. If MaxLen is configured and greater than 0, it trims instead of deleting. Otherwise it DELs the stream key.

func (*Broker) IsConnected

func (b *Broker) IsConnected() bool

IsConnected reports whether the broker is currently connected.

func (*Broker) Metrics

func (b *Broker) Metrics() mq.Metrics

Metrics returns a snapshot of broker metrics.

func (*Broker) Producer

func (b *Broker) Producer(stream string, opts mq.PublishOptions) (mq.Producer, error)

Producer creates or returns a cached producer for the given stream.

func (*Broker) Unbind

func (b *Broker) Unbind(queue, exchange, routingKey string) error

Unbind is a no-op. Redis Streams do not have bindings.

type Config

type Config struct {
	// Addr is the Redis server address (host:port).
	// Example: "localhost:6379".
	Addr string

	// Password is the Redis AUTH password. Empty means no auth.
	Password string

	// DB is the Redis logical database index (0-15).
	DB int

	// Group is the default consumer group name used when a consumer is
	// created in group mode. If empty, consumers run in standalone mode
	// (XREAD) unless overridden per-consumer via ConsumeOptions.Args.
	Group string

	// ConsumerName is the default consumer name within the group.
	// If empty, a unique name is generated per consumer.
	ConsumerName string

	// BlockTime is how long XREADGROUP/XREAD blocks waiting for new
	// messages before retrying. Default: 5s.
	BlockTime time.Duration

	// Count is the maximum number of messages read per XREADGROUP/XREAD
	// call. Default: 10.
	Count int64

	// MaxLen is the maximum stream length. If > 0, every XADD trims the
	// stream to approximately this length (MAXLEN ~). 0 means no trimming.
	MaxLen int64
}

Config configures a Redis Streams broker.

func DefaultConfig

func DefaultConfig() Config

DefaultConfig returns a Config with sensible defaults for localhost.

type Consumer

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

Consumer implements mq.Consumer for Redis Streams.

If a consumer group is configured (broker Group config or per-consumer override), the consumer uses XREADGROUP to read messages assigned to it by the group. Otherwise it uses XREAD in standalone mode.

func (*Consumer) IsRunning

func (c *Consumer) IsRunning() bool

IsRunning reports whether the consumer is actively consuming.

func (*Consumer) Start

func (c *Consumer) Start(ctx context.Context) error

Start begins consuming from the stream. The consumer runs until ctx is cancelled or Stop is called.

func (*Consumer) Stop

func (c *Consumer) Stop(timeout time.Duration) error

Stop gracefully stops consuming, waiting for in-flight handlers to complete up to the given timeout.

type Producer

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

Producer implements mq.Producer for Redis Streams using XADD.

func (*Producer) Close

func (p *Producer) Close() error

Close releases producer resources. After Close, Publish returns ErrClosed.

func (*Producer) Publish

func (p *Producer) Publish(ctx context.Context, msg *mq.Message) error

Publish sends a message to the Redis stream via XADD.

The mq.Message fields are mapped to stream fields as follows:

• "body" -> msg.Body (string)
• "id" -> msg.ID (if set)
• "headers" -> msg.Headers (each key prefixed with "h_")
• metadata fields (content_type, correlation_id, reply_to, type, user_id, app_id) are stored as individual stream fields.

If the broker's MaxLen config is > 0, the stream is trimmed to approximately MaxLen entries on every publish (XADD ... MAXLEN ~ N).