Redis PubSub Support #2610

coolwednesday · 2025-12-01T06:56:04Z

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

Start Publisher (using gofr/examples/using-publisher):
```
cd gofr/examples/using-publisher
go run main.go
```

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"}'

Start Subscriber (using gofr/examples/using-subscriber):
```
cd gofr/examples/using-subscriber
go run main.go
```
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

Start Subscriber first:

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

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

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"}'

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.

pkg/gofr/datasource/pubsub/redis/redis.go

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 · 2025-12-08T11:46:28Z

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

pkg/gofr/datasource/redis/redis.go

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.

Umang01-hash · 2025-12-12T10:03:44Z

pkg/gofr/container/container.go

 	"gofr.dev/pkg/gofr/datasource/pubsub/kafka"
 	"gofr.dev/pkg/gofr/datasource/pubsub/mqtt"
-	"gofr.dev/pkg/gofr/datasource/redis"
+	gofrRedis "gofr.dev/pkg/gofr/datasource/redis"


Any specific reason for adding this alias??

Umang01-hash · 2025-12-12T10:22:16Z

pkg/gofr/container/container.go

+		// Use the embedded PubSub from the Redis client
+		if c.Redis != nil {
+			// Type assert to access PubSub field
+			if redisClient, ok := c.Redis.(*gofrRedis.Redis); ok && redisClient != nil && redisClient.PubSub != nil {


We are not using type assertion for initilalizing any other PubSub client then why do it for redis as it violates the existing architectural pattern. Maybe we can create separate redis.NewPubSub() constructor that returns pubsub.Client directly.

Umang01-hash · 2025-12-12T10:36:46Z

pkg/gofr/datasource/redis/config_test.go

+	}
+}
+
+func TestGetRedisConfig_PubSubStreams_InvalidValues(t *testing.T) {


A common thing that all test methods have is they need a mockLogger and mockConfig, so can we add a testGetRedisConfig(t, configMap) helper function that can take the map and directly return us *Config from getRedisConfig

Umang01-hash · 2025-12-12T10:39:37Z

pkg/gofr/datasource/redis/pubsub.go

+)
+
+const (
+	modePubSub  = "pubsub"


Where this constant is being used??

Umang01-hash · 2025-12-12T10:57:01Z

pkg/gofr/datasource/redis/pubsub.go

+}
+
+func (m *streamMessage) Commit() {
+	err := m.client.XAck(context.Background(), m.stream, m.group, m.id).Err()


Using context.Background() here could cause the Commit operation to hang indefinitely if Redis is slow or unresponsive, consider adding a timeout context to prevent goroutine hangs.

Umang01-hash · 2025-12-12T11:13:07Z

pkg/gofr/datasource/redis/redis.go

+
+	// PubSub constants.
+	defaultRetryTimeout  = 10 * time.Second
+	messageBufferSize    = 100


The message buffer size of 100 is too small for production workloads and will lead to frequent message drops under moderate load. Consider a typical scenario:

Message rate: 500 msg/sec per topic

Consumer processing time: 50ms per message

Buffer fills in: 0.2 seconds

Umang01-hash · 2025-12-12T11:14:01Z

pkg/gofr/datasource/redis/pubsub.go

+
+	hostname, _ := os.Hostname()
+
+	return fmt.Sprintf("consumer-%s-%d", hostname, time.Now().UnixNano())


The consumer name generation using time.Now().UnixNano() can produce duplicate values when called in rapid succession (within the same nanosecond), leading to consumer ID collisions in Redis Streams consumer groups.

Umang01-hash · 2025-12-12T11:18:55Z

pkg/gofr/datasource/redis/pubsub.go

+	}
+
+	if ps.parent.logger != nil {
+		traceID := span.SpanContext().TraceID().String()


traceID is calculated but never used (it's passed to logPubSub() via span, not directly).

Umang01-hash · 2025-12-12T11:23:55Z

pkg/gofr/datasource/redis/pubsub.go

+	}
+
+	addr := fmt.Sprintf("%s:%d", ps.parent.config.HostName, ps.parent.config.Port)
+	res.Details["addr"] = sanitizeRedisAddr(addr)


The sanitizeRedisAddr() function is designed to remove credentials from Redis connection URLs (e.g., redis://user:pass@host:port), but here we're passing a simple host:port string that never contains credentials.

Umang01-hash · 2025-12-12T11:25:36Z

pkg/gofr/datasource/redis/pubsub_test.go

+	require.NoError(t, err)
+
+	// Miniredis compatibility check
+	if len(res) == 0 {


We should not have if else conditions in test? Please remvoe it and separate it to two test methods if requried.

coolwednesday added 7 commits November 25, 2025 16:34

added redis implementation

4c0cc96

refactoring

3d08315

refactoring

1964381

reverting go.mod changes

c741a83

Merge remote-tracking branch 'origin/development' into redis-pubsub

fdcaf90

Adding module in go.work

dfb262c

removed unnecessary added documentation

2a8c55d

github-advanced-security bot found potential problems Dec 1, 2025

View reviewed changes

added tests

90b4434

manavkush reviewed Dec 7, 2025

View reviewed changes

coolwednesday closed this Dec 8, 2025

coolwednesday added 2 commits December 9, 2025 12:10

fixed security issues and linting

ce99c3d

moved to main gofr module & added tests and fixed linters

e3a58a8

coolwednesday reopened this Dec 11, 2025

github-advanced-security bot found potential problems Dec 11, 2025

View reviewed changes

pkg/gofr/datasource/redis/redis.go Fixed Show fixed Hide fixed

pkg/gofr/datasource/redis/redis.go Fixed Show fixed Hide fixed

pkg/gofr/datasource/redis/redis.go Fixed Show fixed Hide fixed

coolwednesday and others added 7 commits December 11, 2025 17:19

reverted unecessary changes

718d16d

Merge branch 'development' into redis-pubsub

258bb62

fixed security issues and linters

93032bc

Merge remote-tracking branch 'origin/redis-pubsub' into redis-pubsub

908a6d5

updating the docs and tests to show default method as streams

1e2100b

fixed small tests

9d6782c

Merge branch 'development' into redis-pubsub

45ed0e4

Umang01-hash reviewed Dec 12, 2025

View reviewed changes


		hostname, _ := os.Hostname()

		return fmt.Sprintf("consumer-%s-%d", hostname, time.Now().UnixNano())

Redis PubSub Support #2610

Are you sure you want to change the base?

Redis PubSub Support #2610

Uh oh!

Conversation

coolwednesday commented Dec 1, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Redis PubSub: Streams Support and Default Mode Change

Testing Redis PubSub

Prerequisites

Testing Streams Mode (Persistence) - Default

Configuration

Test Persistence

Testing PubSub Mode (Fire-and-Forget)

Configuration

Test Fire-and-Forget Behavior

Features

✅ Streams Mode (Default)

✅ PubSub Mode

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

manavkush left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

coolwednesday commented Dec 8, 2025

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Umang01-hash left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

coolwednesday commented Dec 1, 2025 •

edited

Loading

manavkush left a comment •

edited

Loading