fix: random nil dereference caused by non zero PipelineMultiplex (#427)

mux.go

@@ -117,7 +117,7 @@ func (m *mux) SetOnCloseHook(fn func(error)) {
 }
 
 func (m *mux) setCloseHookOnWire(i uint16, w wire) {
-	if w != m.dead {
+	if w != m.dead && w != m.init {
 		w.SetOnCloseHook(func(err error) {
 			if err != ErrClosing {
 				if m.wire[i].CompareAndSwap(w, m.init) {

redis_test.go

@@ -790,8 +790,9 @@ func TestSingleClientIntegration(t *testing.T) {
 	}
 	defer ShouldNotLeaked(SetupLeakDetection())
 	client, err := NewClient(ClientOption{
-		InitAddress:      []string{"127.0.0.1:6379"},
-		ConnWriteTimeout: 180 * time.Second,
+		InitAddress:       []string{"127.0.0.1:6379"},
+		ConnWriteTimeout:  180 * time.Second,
+		PipelineMultiplex: 1,
 	})
 	if err != nil {
 		t.Fatal(err)
@@ -814,7 +815,8 @@ func TestSentinelClientIntegration(t *testing.T) {
 		Sentinel: SentinelOption{
 			MasterSet: "test",
 		},
-		SelectDB: 2, // https://github.com/redis/rueidis/issues/138
+		SelectDB:          2, // https://github.com/redis/rueidis/issues/138
+		PipelineMultiplex: 1,
 	})
 	if err != nil {
 		t.Fatal(err)
@@ -832,10 +834,11 @@ func TestClusterClientIntegration(t *testing.T) {
 	}
 	defer ShouldNotLeaked(SetupLeakDetection())
 	client, err := NewClient(ClientOption{
-		InitAddress:      []string{"127.0.0.1:7001", "127.0.0.1:7002", "127.0.0.1:7003"},
-		ConnWriteTimeout: 180 * time.Second,
-		ShuffleInit:      true,
-		Dialer:           net.Dialer{KeepAlive: -1},
+		InitAddress:       []string{"127.0.0.1:7001", "127.0.0.1:7002", "127.0.0.1:7003"},
+		ConnWriteTimeout:  180 * time.Second,
+		ShuffleInit:       true,
+		Dialer:            net.Dialer{KeepAlive: -1},
+		PipelineMultiplex: 1,
 	})
 	if err != nil {
 		t.Fatal(err)
@@ -851,9 +854,10 @@ func TestSingleClient5Integration(t *testing.T) {
 	}
 	defer ShouldNotLeaked(SetupLeakDetection())
 	client, err := NewClient(ClientOption{
-		InitAddress:      []string{"127.0.0.1:6355"},
-		ConnWriteTimeout: 180 * time.Second,
-		DisableCache:     true,
+		InitAddress:       []string{"127.0.0.1:6355"},
+		ConnWriteTimeout:  180 * time.Second,
+		DisableCache:      true,
+		PipelineMultiplex: 1,
 	})
 	if err != nil {
 		t.Fatal(err)
@@ -870,11 +874,12 @@ func TestCluster5ClientIntegration(t *testing.T) {
 	}
 	defer ShouldNotLeaked(SetupLeakDetection())
 	client, err := NewClient(ClientOption{
-		InitAddress:      []string{"127.0.0.1:7004", "127.0.0.1:7005", "127.0.0.1:7006"},
-		ConnWriteTimeout: 180 * time.Second,
-		ShuffleInit:      true,
-		DisableCache:     true,
-		Dialer:           net.Dialer{KeepAlive: -1},
+		InitAddress:       []string{"127.0.0.1:7004", "127.0.0.1:7005", "127.0.0.1:7006"},
+		ConnWriteTimeout:  180 * time.Second,
+		ShuffleInit:       true,
+		DisableCache:      true,
+		Dialer:            net.Dialer{KeepAlive: -1},
+		PipelineMultiplex: 1,
 	})
 	if err != nil {
 		t.Fatal(err)
@@ -896,6 +901,7 @@ func TestSentinel5ClientIntegration(t *testing.T) {
 		Sentinel: SentinelOption{
 			MasterSet: "test5",
 		},
+		PipelineMultiplex: 1,
 	})
 	if err != nil {
 		t.Fatal(err)
@@ -912,8 +918,9 @@ func TestKeyDBSingleClientIntegration(t *testing.T) {
 	}
 	defer ShouldNotLeaked(SetupLeakDetection())
 	client, err := NewClient(ClientOption{
-		InitAddress:      []string{"127.0.0.1:6344"},
-		ConnWriteTimeout: 180 * time.Second,
+		InitAddress:       []string{"127.0.0.1:6344"},
+		ConnWriteTimeout:  180 * time.Second,
+		PipelineMultiplex: 1,
 	})
 	if err != nil {
 		t.Fatal(err)