Make subscriber close safe to concurrently close (#44)

Signed-off-by: GitHub Actions Bot <[email protected]>
Co-authored-by: GitHub Actions Bot <[email protected]>

Author: jjcollinge

pkg/sql/subscriber.go

@@ -6,6 +6,7 @@ import (
 	"errors"
 	"fmt"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/jackc/pgx/v5"
@@ -115,7 +116,7 @@ type Subscriber struct {
 
 	subscribeWg *sync.WaitGroup
 	closing     chan struct{}
-	closed      bool
+	closed      uint32
 
 	logger watermill.LoggerAdapter
 }
@@ -167,7 +168,7 @@ func newSubscriberID() ([]byte, string, error) {
 }
 
 func (s *Subscriber) Subscribe(ctx context.Context, topic string) (o <-chan *message.Message, err error) {
-	if s.closed {
+	if atomic.LoadUint32(&s.closed) == 1 {
 		return nil, ErrSubscriberClosed
 	}
 
@@ -486,12 +487,10 @@ ResendLoop:
 }
 
 func (s *Subscriber) Close() error {
-	if s.closed {
+	if !atomic.CompareAndSwapUint32(&s.closed, 0, 1) {
 		return nil
 	}
 
-	s.closed = true
-
 	close(s.closing)
 	s.subscribeWg.Wait()