[pubsub] fix lint issues and improve errors model

Author: capcom6

pkg/pubsub/memory.go

@@ -7,7 +7,7 @@ import (
 	"github.com/google/uuid"
 )
 
-type memoryPubSub struct {
+type MemoryPubSub struct {
 	bufferSize uint
 
 	wg      sync.WaitGroup
@@ -21,15 +21,17 @@ type subscriber struct {
 	ctx context.Context
 }
 
-func NewMemory(opts ...Option) *memoryPubSub {
+func NewMemory(opts ...Option) *MemoryPubSub {
 	o := options{
 		bufferSize: 0,
 	}
 	o.apply(opts...)
 
-	return &memoryPubSub{
+	return &MemoryPubSub{
 		bufferSize: o.bufferSize,
 
+		wg:      sync.WaitGroup{},
+		mu:      sync.RWMutex{},
 		topics:  make(map[string]map[string]subscriber),
 		closeCh: make(chan struct{}),
 	}
@@ -38,7 +40,7 @@ func NewMemory(opts ...Option) *memoryPubSub {
 // Publish sends a message to all subscribers of the given topic.
 // This method blocks until all subscribers have received the message
 // or until ctx is cancelled or the pubsub instance is closed.
-func (m *memoryPubSub) Publish(ctx context.Context, topic string, data []byte) error {
+func (m *MemoryPubSub) Publish(ctx context.Context, topic string, data []byte) error {
 	select {
 	case <-m.closeCh:
 		return ErrPubSubClosed
@@ -82,7 +84,7 @@ func (m *memoryPubSub) Publish(ctx context.Context, topic string, data []byte) e
 	return nil
 }
 
-func (m *memoryPubSub) Subscribe(ctx context.Context, topic string) (*Subscription, error) {
+func (m *MemoryPubSub) Subscribe(ctx context.Context, topic string) (*Subscription, error) {
 	select {
 	case <-m.closeCh:
 		return nil, ErrPubSubClosed
@@ -116,7 +118,7 @@ func (m *memoryPubSub) Subscribe(ctx context.Context, topic string) (*Subscripti
 	return &Subscription{id: id, ctx: subCtx, cancel: cancel, ch: ch}, nil
 }
 
-func (m *memoryPubSub) subscribe(id, topic string, sub subscriber) {
+func (m *MemoryPubSub) subscribe(id, topic string, sub subscriber) {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 
@@ -128,7 +130,7 @@ func (m *memoryPubSub) subscribe(id, topic string, sub subscriber) {
 	subscriptions[id] = sub
 }
 
-func (m *memoryPubSub) unsubscribe(id, topic string) {
+func (m *MemoryPubSub) unsubscribe(id, topic string) {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 
@@ -142,7 +144,7 @@ func (m *memoryPubSub) unsubscribe(id, topic string) {
 	}
 }
 
-func (m *memoryPubSub) Close() error {
+func (m *MemoryPubSub) Close() error {
 	select {
 	case <-m.closeCh:
 		return nil
@@ -155,4 +157,4 @@ func (m *memoryPubSub) Close() error {
 	return nil
 }
 
-var _ PubSub = (*memoryPubSub)(nil)
+var _ PubSub = (*MemoryPubSub)(nil)

pkg/pubsub/options.go

@@ -6,12 +6,10 @@ type options struct {
 	bufferSize uint
 }
 
-func (o *options) apply(opts ...Option) *options {
+func (o *options) apply(opts ...Option) {
 	for _, opt := range opts {
 		opt(o)
 	}
-
-	return o
 }
 
 func WithBufferSize(bufferSize uint) Option {

pkg/pubsub/pubsub.go

@@ -6,8 +6,9 @@ import (
 )
 
 var (
-	ErrPubSubClosed = errors.New("pubsub is closed")
-	ErrInvalidTopic = errors.New("invalid topic name")
+	ErrInvalidConfig = errors.New("invalid config")
+	ErrPubSubClosed  = errors.New("pubsub is closed")
+	ErrInvalidTopic  = errors.New("invalid topic name")
 )
 
 type Message struct {

pkg/pubsub/redis.go

@@ -26,7 +26,7 @@ type RedisConfig struct {
 	Prefix string
 }
 
-type redisPubSub struct {
+type RedisPubSub struct {
 	prefix     string
 	bufferSize uint
 
@@ -39,13 +39,13 @@ type redisPubSub struct {
 	closeCh     chan struct{}
 }
 
-func NewRedis(config RedisConfig, opts ...Option) (*redisPubSub, error) {
+func NewRedis(config RedisConfig, opts ...Option) (*RedisPubSub, error) {
 	if config.Prefix != "" && !strings.HasSuffix(config.Prefix, ":") {
 		config.Prefix += ":"
 	}
 
 	if config.Client == nil && config.URL == "" {
-		return nil, fmt.Errorf("no redis client or url provided")
+		return nil, fmt.Errorf("%w: no redis client or url provided", ErrInvalidConfig)
 	}
 
 	client := config.Client
@@ -63,19 +63,21 @@ func NewRedis(config RedisConfig, opts ...Option) (*redisPubSub, error) {
 	}
 	o.apply(opts...)
 
-	return &redisPubSub{
+	return &RedisPubSub{
 		prefix:     config.Prefix,
 		bufferSize: o.bufferSize,
 
 		client:      client,
 		ownedClient: config.Client == nil,
 
+		wg:          sync.WaitGroup{},
+		mu:          sync.Mutex{},
 		subscribers: make(map[string]context.CancelFunc),
 		closeCh:     make(chan struct{}),
 	}, nil
 }
 
-func (r *redisPubSub) Publish(ctx context.Context, topic string, data []byte) error {
+func (r *RedisPubSub) Publish(ctx context.Context, topic string, data []byte) error {
 	select {
 	case <-r.closeCh:
 		return ErrPubSubClosed
@@ -86,10 +88,14 @@ func (r *redisPubSub) Publish(ctx context.Context, topic string, data []byte) er
 		return ErrInvalidTopic
 	}
 
-	return r.client.Publish(ctx, r.prefix+topic, data).Err()
+	if err := r.client.Publish(ctx, r.prefix+topic, data).Err(); err != nil {
+		return fmt.Errorf("failed to publish message: %w", err)
+	}
+
+	return nil
 }
 
-func (r *redisPubSub) Subscribe(ctx context.Context, topic string) (*Subscription, error) {
+func (r *RedisPubSub) Subscribe(ctx context.Context, topic string) (*Subscription, error) {
 	select {
 	case <-r.closeCh:
 		return nil, ErrPubSubClosed
@@ -104,7 +110,7 @@ func (r *redisPubSub) Subscribe(ctx context.Context, topic string) (*Subscriptio
 	_, err := ps.Receive(ctx)
 	if err != nil {
 		closeErr := ps.Close()
-		return nil, errors.Join(fmt.Errorf("can't subscribe: %w", err), closeErr)
+		return nil, errors.Join(fmt.Errorf("failed to subscribe: %w", err), closeErr)
 	}
 
 	id := uuid.NewString()
@@ -160,7 +166,7 @@ func (r *redisPubSub) Subscribe(ctx context.Context, topic string) (*Subscriptio
 	return &Subscription{id: id, ctx: subCtx, cancel: cancel, ch: ch}, nil
 }
 
-func (r *redisPubSub) Close() error {
+func (r *RedisPubSub) Close() error {
 	select {
 	case <-r.closeCh:
 		return nil
@@ -171,10 +177,12 @@ func (r *redisPubSub) Close() error {
 	r.wg.Wait()
 
 	if r.ownedClient {
-		return r.client.Close()
+		if err := r.client.Close(); err != nil {
+			return fmt.Errorf("failed to close redis client: %w", err)
+		}
 	}
 
 	return nil
 }
 
-var _ PubSub = (*redisPubSub)(nil)
+var _ PubSub = (*RedisPubSub)(nil)