fix: close connections that are still blocked while recycling them to…

… pools (#667)

Signed-off-by: Rueian <[email protected]>

pipe.go

@@ -741,7 +741,9 @@ func (p *pipe) Receive(ctx context.Context, subscribe Completed, fn func(message
 }
 
 func (p *pipe) CleanSubscriptions() {
-	if atomic.LoadInt32(&p.state) == 1 {
+	if atomic.LoadInt32(&p.blcksig) != 0 {
+		p.Close()
+	} else if atomic.LoadInt32(&p.state) == 1 {
 		if p.version >= 7 {
 			p.DoMulti(context.Background(), cmds.UnsubscribeCmd, cmds.PUnsubscribeCmd, cmds.SUnsubscribeCmd, cmds.DiscardCmd)
 		} else {

pipe_test.go

@@ -4285,6 +4285,23 @@ func TestPipe_CleanSubscriptions_6(t *testing.T) {
 	)
 }
 
+func TestPipe_CleanSubscriptions_Blocking(t *testing.T) {
+	defer ShouldNotLeaked(SetupLeakDetection())
+	p, mock, cancel, _ := setup(t, ClientOption{ConnWriteTimeout: time.Second / 2, Dialer: net.Dialer{KeepAlive: time.Second / 3}})
+	defer cancel()
+	p.background()
+	ctx, cancel := context.WithCancel(context.Background())
+	go func() {
+		mock.Expect("BLPOP")
+		cancel()
+	}()
+	p.Do(ctx, cmds.NewBlockingCompleted([]string{"BLPOP"}))
+	p.CleanSubscriptions()
+	if p.Error() != ErrClosing {
+		t.Fatal("unexpected error")
+	}
+}
+
 func TestPipe_CleanSubscriptions_7(t *testing.T) {
 	defer ShouldNotLeaked(SetupLeakDetection())
 	p, mock, cancel, _ := setup(t, ClientOption{ConnWriteTimeout: time.Second / 2, Dialer: net.Dialer{KeepAlive: time.Second / 3}})