cleaning up

Author: coolwednesday

pkg/gofr/datasource/redis/config_test.go

@@ -13,8 +13,6 @@ import (
 	"gofr.dev/pkg/gofr/logging"
 )
 
-// testGetRedisConfig is a helper function that creates mock logger and config,
-// then returns the Redis config. This reduces code duplication in tests.
 func testGetRedisConfig(t *testing.T, configMap map[string]string) *Config {
 	t.Helper()
 
@@ -51,7 +49,6 @@ func TestGetRedisConfig_InvalidPortAndDB(t *testing.T) {
 }
 
 func TestGetRedisConfig_TLS(t *testing.T) {
-	// Create temporary cert files
 	certFile, err := os.CreateTemp(t.TempDir(), "cert-*.pem")
 	require.NoError(t, err)
 
@@ -70,12 +67,10 @@ func TestGetRedisConfig_TLS(t *testing.T) {
 	defer os.Remove(caFile.Name())
 	defer caFile.Close()
 
-	// Write dummy content (not valid PEM, but enough to trigger file read)
 	_, _ = certFile.WriteString("-----BEGIN CERTIFICATE-----\nMIID\n-----END CERTIFICATE-----")
 	_, _ = keyFile.WriteString("-----BEGIN PRIVATE KEY-----\nMIIE\n-----END PRIVATE KEY-----")
 	_, _ = caFile.WriteString("-----BEGIN CERTIFICATE-----\nMIID\n-----END CERTIFICATE-----")
 
-	// This will log errors because dummy content is not valid PEM, but it tests the path
 	conf := testGetRedisConfig(t, map[string]string{
 		"REDIS_HOST":        "localhost",
 		"REDIS_TLS_ENABLED": "true",
@@ -98,7 +93,6 @@ func TestGetRedisConfig_TLS_InvalidFiles(t *testing.T) {
 	})
 
 	assert.NotNil(t, conf.TLS)
-	// Should be empty as files failed to load
 	assert.Empty(t, conf.TLS.Certificates)
 	assert.Nil(t, conf.TLS.RootCAs)
 }
@@ -153,7 +147,6 @@ func TestGetRedisConfig_PubSubStreams_InvalidValues(t *testing.T) {
 	assert.Equal(t, "streams", conf.PubSubMode)
 	require.NotNil(t, conf.PubSubStreamsConfig)
 
-	// Should use defaults
 	assert.Equal(t, int64(0), conf.PubSubStreamsConfig.MaxLen)
 	assert.Equal(t, 5*time.Second, conf.PubSubStreamsConfig.Block)
 }

pkg/gofr/datasource/redis/health.go

@@ -19,7 +19,6 @@ const (
 func (r *Redis) HealthCheck() datasource.Health {
 	h := datasource.Health{
 		Details: make(map[string]any),
-		Status:  datasource.StatusDown,
 	}
 
 	h.Details["host"] = r.config.HostName + ":" + strconv.Itoa(r.config.Port)

pkg/gofr/datasource/redis/health_test.go

@@ -21,7 +21,6 @@ func TestPubSub_HealthDown(t *testing.T) {
 	client, mock := setupMockTest(t, nil)
 	defer client.Close()
 
-	// Test Health Down (Ping fails)
 	mock.ExpectPing().SetErr(errMockPing)
 
 	h := client.PubSub.Health()
@@ -35,7 +34,6 @@ func TestPubSub_HealthUp(t *testing.T) {
 	client, mock := setupMockTest(t, nil)
 	defer client.Close()
 
-	// Test Health Up (Ping succeeds)
 	mock.ExpectPing().SetVal("PONG")
 
 	h := client.PubSub.Health()
@@ -68,7 +66,6 @@ func TestPubSub_HealthDefaultMode(t *testing.T) {
 	client, mock := setupMockTest(t, map[string]string{
 		"REDIS_HOST": "localhost",
 		"REDIS_PORT": "6379",
-		// REDIS_PUBSUB_MODE not set, should default to streams
 	})
 	defer client.Close()
 

pkg/gofr/datasource/redis/messages_test.go

@@ -27,7 +27,6 @@ func TestPubSubMessage_Commit(t *testing.T) {
 	ctx := context.Background()
 	topic := "pubsub-commit-topic"
 
-	// Start subscription first to ensure it's ready
 	msgChan := make(chan *pubsub.Message)
 	errChan := make(chan error)
 
@@ -41,22 +40,18 @@ func TestPubSubMessage_Commit(t *testing.T) {
 		msgChan <- msg
 	}()
 
-	// Wait a bit for subscription to be ready
 	time.Sleep(100 * time.Millisecond)
 
-	// Then publish the message
 	err := client.PubSub.Publish(ctx, topic, []byte("test"))
 	require.NoError(t, err)
 
-	// Wait for message with timeout
 	select {
 	case err := <-errChan:
 		require.NoError(t, err)
 	case msg := <-msgChan:
 		require.NotNil(t, msg)
 		require.NotNil(t, msg.Committer)
 
-		// Commit should not panic
 		assert.NotPanics(t, func() {
 			msg.Committer.Commit()
 		})
@@ -163,7 +158,6 @@ func TestStreamMessage_Commit_Success(t *testing.T) {
 
 	mock.ExpectXAck(stream, group, id).SetVal(1)
 
-	// Test stream message commit through actual stream subscription
 	client, s := setupTest(t, map[string]string{
 		"REDIS_PUBSUB_MODE":            "streams",
 		"REDIS_STREAMS_CONSUMER_GROUP": group,
@@ -173,21 +167,18 @@ func TestStreamMessage_Commit_Success(t *testing.T) {
 
 	ctx := context.Background()
 
-	// Create topic and publish
 	err := client.PubSub.CreateTopic(ctx, stream)
 	require.NoError(t, err)
 
 	go func() {
 		_ = client.PubSub.Publish(ctx, stream, []byte("test"))
 	}()
 
-	// Subscribe to get message with committer
 	msg, err := client.PubSub.Subscribe(ctx, stream)
 	require.NoError(t, err)
 	require.NotNil(t, msg)
 	require.NotNil(t, msg.Committer)
 
-	// Commit should not panic
 	assert.NotPanics(t, func() {
 		msg.Committer.Commit()
 	})
@@ -196,7 +187,6 @@ func TestStreamMessage_Commit_Success(t *testing.T) {
 func TestStreamMessage_Commit_Error(t *testing.T) {
 	t.Parallel()
 
-	// Test stream message commit error handling
 	client, s := setupTest(t, map[string]string{
 		"REDIS_PUBSUB_MODE":            "streams",
 		"REDIS_STREAMS_CONSUMER_GROUP": "test-group",
@@ -207,24 +197,20 @@ func TestStreamMessage_Commit_Error(t *testing.T) {
 	ctx := context.Background()
 	stream := "test-stream-error"
 
-	// Create topic and publish
 	err := client.PubSub.CreateTopic(ctx, stream)
 	require.NoError(t, err)
 
 	go func() {
 		_ = client.PubSub.Publish(ctx, stream, []byte("test"))
 	}()
 
-	// Subscribe to get message
 	msg, err := client.PubSub.Subscribe(ctx, stream)
 	require.NoError(t, err)
 	require.NotNil(t, msg)
 	require.NotNil(t, msg.Committer)
 
-	// Close Redis to simulate error
 	s.Close()
 
-	// Commit should handle error gracefully
 	assert.NotPanics(t, func() {
 		msg.Committer.Commit()
 	})
@@ -233,7 +219,6 @@ func TestStreamMessage_Commit_Error(t *testing.T) {
 func TestStreamMessage_Commit_Timeout(t *testing.T) {
 	t.Parallel()
 
-	// Test stream message commit with timeout
 	client, s := setupTest(t, map[string]string{
 		"REDIS_PUBSUB_MODE":            "streams",
 		"REDIS_STREAMS_CONSUMER_GROUP": "test-group",
@@ -244,21 +229,18 @@ func TestStreamMessage_Commit_Timeout(t *testing.T) {
 	ctx := context.Background()
 	stream := "test-stream-timeout"
 
-	// Create topic and publish
 	err := client.PubSub.CreateTopic(ctx, stream)
 	require.NoError(t, err)
 
 	go func() {
 		_ = client.PubSub.Publish(ctx, stream, []byte("test"))
 	}()
 
-	// Subscribe to get message
 	msg, err := client.PubSub.Subscribe(ctx, stream)
 	require.NoError(t, err)
 	require.NotNil(t, msg)
 	require.NotNil(t, msg.Committer)
 
-	// Commit should handle timeout gracefully (uses defaultRetryTimeout internally)
 	assert.NotPanics(t, func() {
 		msg.Committer.Commit()
 	})

pkg/gofr/datasource/redis/pubsub.go

@@ -280,10 +280,6 @@ func (ps *PubSub) ensureConsumerGroup(ctx context.Context, topic, group string)
 
 // checkGroupExists checks if a consumer group exists for the given stream.
 func (ps *PubSub) checkGroupExists(ctx context.Context, topic, group string) bool {
-	if !ps.isConnected() {
-		return false
-	}
-
 	groups, err := ps.client.XInfoGroups(ctx, topic).Result()
 	if err != nil {
 		// If XInfoGroups failed (e.g., stream doesn't exist), we'll create it with MKSTREAM
@@ -460,7 +456,6 @@ func (ps *PubSub) createStreamTopic(ctx context.Context, name string) error {
 
 	group := ps.config.PubSubStreamsConfig.ConsumerGroup
 
-	// checkGroupExists already calls isConnected()
 	groupExists := ps.checkGroupExists(ctx, name, group)
 	if groupExists {
 		return nil
@@ -734,14 +729,14 @@ func (*PubSub) collectMessages(ctx context.Context, ch <-chan *redis.Message, li
 	collected := 0
 
 	for collected < limit {
-		// Check context first
+		// Check context first before attempting to receive from channel
 		select {
 		case <-ctx.Done():
 			return result
 		default:
 		}
 
-		// Then try to receive message
+		// Now try to receive from channel, but also check context
 		select {
 		case <-ctx.Done():
 			return result