3021_add_subscriptionMap_log_subscription

Author: ronak4195

extra/rediscensus/go.mod

@@ -19,6 +19,6 @@ require (
 )
 
 retract (
-	v9.5.3 // This version was accidentally released.
 	v9.7.2 // This version was accidentally released.
+	v9.5.3 // This version was accidentally released.
 )

extra/rediscmd/go.mod

@@ -16,6 +16,6 @@ require (
 )
 
 retract (
-	v9.5.3 // This version was accidentally released.
 	v9.7.2 // This version was accidentally released.
+	v9.5.3 // This version was accidentally released.
 )

extra/redisotel/go.mod

@@ -24,6 +24,6 @@ require (
 )
 
 retract (
-	v9.5.3 // This version was accidentally released.
 	v9.7.2 // This version was accidentally released.
+	v9.5.3 // This version was accidentally released.
 )

extra/redisprometheus/go.mod

@@ -23,6 +23,6 @@ require (
 )
 
 retract (
-	v9.5.3 // This version was accidentally released.
 	v9.7.2 // This version was accidentally released.
+	v9.5.3 // This version was accidentally released.
 )

pubsub.go

@@ -3,6 +3,7 @@ package redis
 import (
 	"context"
 	"fmt"
+	"sort"
 	"strings"
 	"sync"
 	"time"
@@ -571,6 +572,9 @@ type channel struct {
 	chanSize        int
 	chanSendTimeout time.Duration
 	checkInterval   time.Duration
+
+	subscriptions   map[string]struct{}
+	subscriptionsMu sync.RWMutex
 }
 
 func newChannel(pubSub *PubSub, opts ...ChannelOption) *channel {
@@ -580,6 +584,8 @@ func newChannel(pubSub *PubSub, opts ...ChannelOption) *channel {
 		chanSize:        100,
 		chanSendTimeout: time.Minute,
 		checkInterval:   3 * time.Second,
+
+		subscriptions: make(map[string]struct{}),
 	}
 	for _, opt := range opts {
 		opt(c)
@@ -618,6 +624,20 @@ func (c *channel) initHealthCheck() {
 	}()
 }
 
+// Helper function to format subscription information
+func (c *channel) getSubscriptionInfo() string {
+	if len(c.subscriptions) == 0 {
+		return "none"
+	}
+
+	subs := make([]string, 0, len(c.subscriptions))
+	for sub := range c.subscriptions {
+		subs = append(subs, sub)
+	}
+	sort.Strings(subs) // Sort for consistent output
+	return strings.Join(subs, ", ")
+}
+
 // initMsgChan must be in sync with initAllChan.
 func (c *channel) initMsgChan() {
 	ctx := context.TODO()
@@ -663,9 +683,13 @@ func (c *channel) initMsgChan() {
 						<-timer.C
 					}
 				case <-timer.C:
+					c.subscriptionsMu.RLock()
+					subInfo := c.getSubscriptionInfo()
+					c.subscriptionsMu.RUnlock()
+
 					internal.Logger.Printf(
-						ctx, "redis: %s channel is full for %s (message is dropped)",
-						c, c.chanSendTimeout)
+						ctx, "redis: Channel is full for %s (message is dropped), subscriptions: %s",
+						c.chanSendTimeout, subInfo)
 				}
 			default:
 				internal.Logger.Printf(ctx, "redis: unknown message type: %T", msg)
@@ -717,9 +741,13 @@ func (c *channel) initAllChan() {
 						<-timer.C
 					}
 				case <-timer.C:
+					c.subscriptionsMu.RLock()
+					subInfo := c.getSubscriptionInfo()
+					c.subscriptionsMu.RUnlock()
+
 					internal.Logger.Printf(
-						ctx, "redis: %s channel is full for %s (message is dropped)",
-						c, c.chanSendTimeout)
+						ctx, "redis: Channel is full for %s (message is dropped), subscriptions: %s",
+						c.chanSendTimeout, subInfo)
 				}
 			default:
 				internal.Logger.Printf(ctx, "redis: unknown message type: %T", msg)