Redis PubSub Support

[coolwednesday]

Redis PubSub: Streams Support and Default Mode Change

Description:

• Closes Redis Pub Sub Support #2347
• This PR adds comprehensive support for Redis Streams in GoFr's PubSub implementation and changes the default mode from pubsub to streams. The implementation follows the same patterns as other pubsub providers (NATS, EventHub, Kafka) and includes full support for consumer groups, acknowledgments, and message persistence.

Testing Redis PubSub

Prerequisites

Start Redis using Docker:

# Basic Redis instance
docker run -d \
  --name redis \
  -p 6379:6379 \
  redis:7-alpine

# Redis with password authentication
docker run -d \
  --name redis \
  -p 6379:6379 \
  redis:7-alpine redis-server --requirepass mypassword

Testing Streams Mode (Persistence) - Default

Streams mode provides persistent messaging with consumer groups and acknowledgments. Messages are stored in Redis and can be processed even after app restarts.

Configuration

Create configs/.env:

PUBSUB_BACKEND=REDIS
REDIS_HOST=localhost
REDIS_PORT=6379
# Streams mode is default, but explicitly set for clarity
REDIS_PUBSUB_MODE=streams
REDIS_STREAMS_CONSUMER_GROUP=my-group
REDIS_STREAMS_CONSUMER_NAME=my-consumer
REDIS_STREAMS_BLOCK_TIMEOUT=5s
REDIS_STREAMS_MAXLEN=1000

Test Persistence

1. Start Publisher (using gofr/examples/using-publisher):

   cd gofr/examples/using-publisher
   go run main.go

2. Publish messages:

   # Publish an order
   curl -X POST http://localhost:8000/publish-order \
     -H "Content-Type: application/json" \
     -d '{"orderId": "order-123", "status": "pending"}'
   
   # Publish a product
   curl -X POST http://localhost:8000/publish-product \
     -H "Content-Type: application/json" \
     -d '{"productId": "prod-456", "price": "99.99"}'

3. Start Subscriber (using gofr/examples/using-subscriber):

   cd gofr/examples/using-subscriber
   go run main.go

4. Verify Persistence:

   • Stop the subscriber
   • Publish more messages
   • Restart the subscriber
   • Expected: Subscriber processes all messages, including those published while it was down
   • Messages are persisted in Redis Streams and can be replayed

Testing PubSub Mode (Fire-and-Forget)

PubSub mode provides non-persistent messaging. Messages are delivered only to active subscribers and are lost if no subscriber is listening.

Configuration

Create configs/.env:

PUBSUB_BACKEND=REDIS
REDIS_HOST=localhost
REDIS_PORT=6379
REDIS_PUBSUB_MODE=pubsub

Test Fire-and-Forget Behavior

1. Start Subscriber first:

   cd gofr/examples/using-subscriber
   go run main.go

2. Publish messages (subscriber is listening):

   curl -X POST http://localhost:8000/publish-order \
     -H "Content-Type: application/json" \
     -d '{"orderId": "order-123", "status": "pending"}'

   • Expected: Subscriber receives and processes the message immediately

3. Stop the subscriber and publish more messages:

   # Stop subscriber (Ctrl+C)
   # Then publish
   curl -X POST http://localhost:8000/publish-order \
     -H "Content-Type: application/json" \
     -d '{"orderId": "order-456", "status": "processing"}'

4. Restart subscriber:

   go run main.go

   • Expected: Subscriber does NOT receive messages published while it was down
   • Messages are lost (fire-and-forget behavior)

Features

✅ Streams Mode (Default)

• Persistence: Messages are stored in Redis Streams
• Consumer Groups: Multiple consumers can process messages
• Acknowledgments: Messages are acknowledged after processing
• At-least-once delivery: Messages are guaranteed to be delivered
• Replay capability: Messages can be replayed from any point

✅ PubSub Mode

• Fire-and-forget: Messages delivered only to active subscribers
• No persistence: Messages are not stored
• At-most-once delivery: Messages may be lost if no subscriber is active
• Lightweight: Lower overhead than streams

Checklist:

• I have formatted my code using goimport and golangci-lint.
• All new code is covered by unit tests.
• This PR does not decrease the overall code coverage.
• I have reviewed the code comments and documentation for clarity.

[manavkush]

Changes seem good 👍 . Didn't test yet though.
@coolwednesday the AI suggestions for the logging can be added for redacting sensitive information. Can check this in other critical flows as well.

[coolwednesday]

Will reopen the PR after fixing the review comments, linters, tests and proper testing.

[Umang01-hash]

• Can we rename the file metrics_interface.go inside redis directory to mock_metrics.go.
• We have a file config_test.go but in respect to that config.go file doesn;t exist. That violates Go naming conventions and makes code harder to discover.
• The file pubsub_query_test.go contains only 2 tests for the Query() method, but pubsub_test.go already has query tests (testChannelQuery() and testStreamQuery() in getQueryTestCases()). We can delete pubsub_query_test.go and merge the 2 tests into pubsub_test.go.