redissub

Overview ¶

Package redissub provides a Redis Stream subscriber with automatic retry, timeout, and lifecycle management.

This package wraps Watermill's Redis Stream subscriber to consume events from Redis Streams with configurable retry logic, execution timeouts, and graceful shutdown. It supports consumer groups for distributed message processing.

Basic usage:

handler := func(ctx context.Context, msg *message.Message) error {
    var event MyEvent
    if err := json.Unmarshal(msg.Payload, &event); err != nil {
        return err
    }
    // process event
    return nil
}

subscriber, err := redissub.New(redisClient, "my-consumer-group", &redissub.Config{
    ExecTimeout:     30 * time.Second,
    MaxRetries:      3,
    ShutdownTimeout: 20 * time.Second,
})
if err != nil {
    log.Fatal(err)
}

subscriber.Subscribe(ctx, "events-topic", handler)
if err != nil {
    log.Fatal(err)
}

Messages are acknowledged only after successful processing. Failed messages are retried up to MaxRetries times before being nacked. Timeouts on message handlers are enforced via the ExecTimeout configuration.