fix: fix panic when redis client closed before server closed

subscriber.go

@@ -46,7 +46,7 @@ func newSubscriber(params subscriberParams) *subscriber {
 func (s *subscriber) shutdown() {
 	s.logger.Debug("Subscriber shutting down...")
 	// Signal the subscriber goroutine to stop.
-	s.done <- struct{}{}
+	close(s.done)
 }
 
 func (s *subscriber) start(wg *sync.WaitGroup) {
@@ -80,6 +80,11 @@ func (s *subscriber) start(wg *sync.WaitGroup) {
 				s.logger.Debug("Subscriber done")
 				return
 			case msg := <-cancelCh:
+				if msg == nil {
+					s.logger.Warn("channel closed")
+					pubsub.Close()
+					return
+				}
 				cancel, ok := s.cancelations.Get(msg.Payload)
 				if ok {
 					cancel()

subscriber_test.go

@@ -70,6 +70,38 @@ func TestSubscriber(t *testing.T) {
 	}
 }
 
+func TestSubscriberWithRedisClientClosed(t *testing.T) {
+	defer func() {
+		if r := recover(); r != nil {
+			t.Errorf("panic occurred: %v", r)
+		}
+	}()
+	redisClient := setup(t)
+	r := rdb.NewRDB(redisClient)
+	defer r.Close()
+	testBroker := testbroker.NewTestBroker(r)
+
+	cancelations := base.NewCancelations()
+	subscriber := newSubscriber(subscriberParams{
+		logger:       testLogger,
+		broker:       testBroker,
+		cancelations: cancelations,
+	})
+	subscriber.retryTimeout = 1 * time.Second // set shorter retry timeout for testing purpose.
+
+	var wg sync.WaitGroup
+	subscriber.start(&wg)
+	defer subscriber.shutdown()
+
+	time.Sleep(5 * time.Second) // allow some time for subscriber to connect
+
+	if err := redisClient.Close(); err != nil {
+		t.Fatalf("close redis client,%+v", err)
+	}
+
+	time.Sleep(1 * time.Second) // make sure client is closed
+}
+
 func TestSubscriberWithRedisDown(t *testing.T) {
 	defer func() {
 		if r := recover(); r != nil {