mqutils
GitLab ↗

Docs / Backends / Redis

Redis

Redis Pub/Sub channels (at-most-once) and Redis Streams consumer groups with XAutoClaim recovery

go
import "go.digitalxero.dev/mq-redis/v2"

Index

funcNewMessageBuilder

go
func NewMessageBuilder() mqtypes.MessageBuilder

NewMessageBuilder creates a new Redis message builder. The builder provides a fluent interface for constructing Redis messages with support for both Pub/Sub and Streams modes.

Redis message considerations:

  • Pub/Sub: Messages are ephemeral, no size limits beyond memory
  • Streams: Messages are persistent with configurable retention
  • Both modes serialize the entire message structure to JSON

Example:

go
msg := redis.NewMessageBuilder().
    WithBody([]byte(`{"action": "process", "itemId": 789}`)).
    WithContentType("application/json").
    WithCorrelationId("task-789").
    WithRoutingKey("*"). // Stream ID (* = auto-generate)
    WithHeaders(map[string]any{
        "priority": 1,
        "retry_count": 0,
        "timestamp": time.Now().Unix(),
    }).
    Build()

funcNewRedisConsumer

go
func NewRedisConsumer(config *viper.Viper) (types.Consumer, error)

NewRedisConsumer creates a new Redis consumer with the provided configuration. This function is typically called by mqutils.NewConsumer when it detects a Redis URL.

Configuration options:

  • url: Redis connection URL (required unless addr is set; schemes redis://, rediss://, redisstream://, redisstreams://)
  • channel_name / stream_name: consume target for Pub/Sub / Streams mode (the canonical “destination” key maps onto whichever is native for the detected mode)
  • use_streams: consume from a Redis Stream instead of Pub/Sub (default: derived from the URL scheme)
  • consumer_group / consumer_name: Redis Streams consumer group identity (stream mode only; “consumer_group” is also the canonical key)
  • skip_verify: skip TLS certificate verification for rediss:// and redisstreams:// URLs (default: false)
  • handler: Name of registered handler function (default: “redisLogger”, or “redisBatchLogger” when batch processing is enabled)
  • password / username / db: Redis authentication and database selection
  • max_retries: delivery retry budget before a message is dropped, also applied as the client’s command retry limit when set (default: 50)
  • min_retry_backoff_ms / max_retry_backoff_ms: client retry backoff bounds (defaults: 8 / 512)
  • dial_timeout_seconds / read_timeout_seconds / write_timeout_seconds: client timeouts (defaults: 5 / 3 / 3)
  • pool_size / min_idle_conns / max_idle_conns / conn_max_lifetime_minutes: connection pool tuning (defaults: 10 / 1 / pool_size / 30)
  • block_time_seconds: XREADGROUP/consume blocking time (default: 1)
  • stream_max_len / stream_trim_approx: XADD MAXLEN trimming applied to stream publishes through this transport (defaults: 10000 / true)
  • claim_idle_time_seconds: minimum idle time before a pending message is reclaimed from a dead consumer via XPENDING/XCLAIM (default: 60; requires Redis >= 6.2)
  • pending_message_max_age_seconds: pending messages idle longer than this are acknowledged and dropped as poison messages (default: 300; 0 disables)
  • claim_interval_seconds: recovery sweep interval (default: derived as claim_idle_time/2, floored at 5s)
  • batch_size: Number of messages per batch (default: 5)
  • batch_timeout: Batch collection window as a Go duration (default: “100ms”)
  • enable_batch_processing: Enable batch message processing (default: false)
  • max_concurrent_handlers: Maximum in-flight handlers or batches (default: effective message_channel_buffer; must be positive)

Typed configuration values take precedence over URL query parameters, which take precedence over defaults.

Returns an error if configuration validation fails or transport creation fails.

funcNewRedisProducer

go
func NewRedisProducer(config *viper.Viper) (types.Producer, error)

NewRedisProducer creates a new Redis producer with the provided configuration. This function is typically called by mqutils.NewProducer when it detects a Redis URL.

Configuration options:

  • url: Redis connection URL (required; schemes redis://, rediss://, redisstream://, redisstreams://)
  • channel / stream_key: publish target for pubsub / stream mode (the canonical “destination” key maps onto whichever is native for the detected mode)
  • mode: Producer mode - “pubsub” or “stream” (default: derived from the URL scheme, falling back to “pubsub”)
  • skip_verify: skip TLS certificate verification for rediss:// and redisstreams:// URLs (default: false)
  • password: Redis password for authentication (optional)
  • database: Redis database number (default: 0)
  • max_len: Max stream length for XADD MAXLEN trimming (stream mode, default: 10000)
  • approximate_max_len: Use approximate ("~") trimming (default: true)

Typed configuration values take precedence over URL query parameters, which take precedence over defaults.

Returns an error if configuration validation fails or transport creation fails. The producer must be started with Start() before publishing messages.

funcNewRedisTransport

go
func NewRedisTransport() mqtypes.Transport

NewRedisTransport creates a new Redis transport implementation. The transport provides low-level Redis operations for both Pub/Sub and Streams modes, with automatic mode detection based on URL scheme.

The transport manages:

  • Redis UniversalClient supporting single node and cluster
  • Connection pooling with configurable limits
  • Automatic reconnection with exponential backoff
  • Mode switching between Pub/Sub and Streams
  • Pending-message recovery via XPENDING/XCLAIM (streams with a consumer group; requires Redis >= 6.2)
  • Health monitoring with server statistics

Supported URL schemes:

  • redis:// - Standard Redis with Pub/Sub mode
  • rediss:// - Redis Pub/Sub with TLS
  • redisstream:// - Redis Streams mode
  • redisstreams:// - Redis Streams with TLS

Experimental: the URL host may be a comma-separated list of host:port addresses (e.g. redis://a:6379,b:6379). With two or more addresses the underlying client operates in cluster mode, where the database number is ignored (Redis Cluster only supports database 0).

Generated by gomarkdoc

↑ ↓ move↵ open/ opens search anywhere