refactor: changed atomic.Value to atomic.Pointer in pipe (#851)

* refactor: changed atomic.Value to atomic.Pointer in pipe

* fix: fixed unit test failure in mock

Author: nithinputhenveettil

mock/result.go

@@ -186,8 +186,8 @@ type pool struct {
 type pipe struct {
 	conn            net.Conn
 	error           atomic.Pointer[errs]
-	clhks           atomic.Value // closed hook, invoked after the conn is closed
-	pshks           atomic.Value // pubsub hook, registered by the SetPubSubHooks
+	clhks           atomic.Value          // closed hook, invoked after the conn is closed
+	pshks           atomic.Pointer[pshks] // pubsub hook, registered by the SetPubSubHooks
 	queue           any
 	cache           rueidis.CacheStore
 	r               *bufio.Reader
@@ -225,3 +225,8 @@ type stream struct {
 }
 
 type errs struct{ error }
+
+type pshks struct {
+	hooks rueidis.PubSubHooks
+	close chan error
+}

pipe.go

@@ -63,8 +63,8 @@ var _ wire = (*pipe)(nil)
 
 type pipe struct {
 	conn            net.Conn
-	clhks           atomic.Value // closed hook, invoked after the conn is closed
-	pshks           atomic.Value // pubsub hook, registered by the SetPubSubHooks
+	clhks           atomic.Value          // closed hook, invoked after the conn is closed
+	pshks           atomic.Pointer[pshks] // pubsub hook, registered by the SetPubSubHooks
 	queue           queue
 	cache           CacheStore
 	error           atomic.Pointer[errs]
@@ -393,7 +393,7 @@ func (p *pipe) _background() {
 	p.nsubs.Close()
 	p.psubs.Close()
 	p.ssubs.Close()
-	if old := p.pshks.Swap(emptypshks).(*pshks); old.close != nil {
+	if old := p.pshks.Swap(emptypshks); old.close != nil {
 		old.close <- err
 		close(old.close)
 	}
@@ -705,60 +705,60 @@ func (p *pipe) handlePush(values []RedisMessage) (reply bool, unsubscribe bool)
 		if len(values) >= 3 {
 			m := PubSubMessage{Channel: values[1].string(), Message: values[2].string()}
 			p.nsubs.Publish(values[1].string(), m)
-			p.pshks.Load().(*pshks).hooks.OnMessage(m)
+			p.pshks.Load().hooks.OnMessage(m)
 		}
 	case "pmessage":
 		if len(values) >= 4 {
 			m := PubSubMessage{Pattern: values[1].string(), Channel: values[2].string(), Message: values[3].string()}
 			p.psubs.Publish(values[1].string(), m)
-			p.pshks.Load().(*pshks).hooks.OnMessage(m)
+			p.pshks.Load().hooks.OnMessage(m)
 		}
 	case "smessage":
 		if len(values) >= 3 {
 			m := PubSubMessage{Channel: values[1].string(), Message: values[2].string()}
 			p.ssubs.Publish(values[1].string(), m)
-			p.pshks.Load().(*pshks).hooks.OnMessage(m)
+			p.pshks.Load().hooks.OnMessage(m)
 		}
 	case "unsubscribe":
 		if len(values) >= 3 {
 			s := PubSubSubscription{Kind: values[0].string(), Channel: values[1].string(), Count: values[2].intlen}
 			p.nsubs.Unsubscribe(s)
-			p.pshks.Load().(*pshks).hooks.OnSubscription(s)
+			p.pshks.Load().hooks.OnSubscription(s)
 		}
 		return true, true
 	case "punsubscribe":
 		if len(values) >= 3 {
 			s := PubSubSubscription{Kind: values[0].string(), Channel: values[1].string(), Count: values[2].intlen}
 			p.psubs.Unsubscribe(s)
-			p.pshks.Load().(*pshks).hooks.OnSubscription(s)
+			p.pshks.Load().hooks.OnSubscription(s)
 		}
 		return true, true
 	case "sunsubscribe":
 		if len(values) >= 3 {
 			s := PubSubSubscription{Kind: values[0].string(), Channel: values[1].string(), Count: values[2].intlen}
 			p.ssubs.Unsubscribe(s)
-			p.pshks.Load().(*pshks).hooks.OnSubscription(s)
+			p.pshks.Load().hooks.OnSubscription(s)
 		}
 		return true, true
 	case "subscribe":
 		if len(values) >= 3 {
 			s := PubSubSubscription{Kind: values[0].string(), Channel: values[1].string(), Count: values[2].intlen}
 			p.nsubs.Confirm(s)
-			p.pshks.Load().(*pshks).hooks.OnSubscription(s)
+			p.pshks.Load().hooks.OnSubscription(s)
 		}
 		return true, false
 	case "psubscribe":
 		if len(values) >= 3 {
 			s := PubSubSubscription{Kind: values[0].string(), Channel: values[1].string(), Count: values[2].intlen}
 			p.psubs.Confirm(s)
-			p.pshks.Load().(*pshks).hooks.OnSubscription(s)
+			p.pshks.Load().hooks.OnSubscription(s)
 		}
 		return true, false
 	case "ssubscribe":
 		if len(values) >= 3 {
 			s := PubSubSubscription{Kind: values[0].string(), Channel: values[1].string(), Count: values[2].intlen}
 			p.ssubs.Confirm(s)
-			p.pshks.Load().(*pshks).hooks.OnSubscription(s)
+			p.pshks.Load().hooks.OnSubscription(s)
 		}
 		return true, false
 	}
@@ -871,7 +871,7 @@ func (p *pipe) SetPubSubHooks(hooks PubSubHooks) <-chan error {
 		return p._r2pipe(context.Background()).SetPubSubHooks(hooks)
 	}
 	if hooks.isZero() {
-		if old := p.pshks.Swap(emptypshks).(*pshks); old.close != nil {
+		if old := p.pshks.Swap(emptypshks); old.close != nil {
 			close(old.close)
 		}
 		return nil
@@ -883,11 +883,11 @@ func (p *pipe) SetPubSubHooks(hooks PubSubHooks) <-chan error {
 		hooks.OnSubscription = func(s PubSubSubscription) {}
 	}
 	ch := make(chan error, 1)
-	if old := p.pshks.Swap(&pshks{hooks: hooks, close: ch}).(*pshks); old.close != nil {
+	if old := p.pshks.Swap(&pshks{hooks: hooks, close: ch}); old.close != nil {
 		close(old.close)
 	}
 	if err := p.Error(); err != nil {
-		if old := p.pshks.Swap(emptypshks).(*pshks); old.close != nil {
+		if old := p.pshks.Swap(emptypshks); old.close != nil {
 			old.close <- err
 			close(old.close)
 		}