Merge pull request #25 from botchris/snssqs-write-only-options

[SNSSQS] Write-only broker option

Author: qfornaguera

provider/snssqs/broker.go

@@ -13,6 +13,7 @@ import (
 	"github.com/aws/aws-sdk-go-v2/service/sns"
 	"github.com/aws/aws-sdk-go-v2/service/sqs"
 	"github.com/botchris/go-pubsub"
+	"github.com/botchris/go-pubsub/middleware/codec"
 	"github.com/hashicorp/go-multierror"
 	"github.com/kubemq-io/kubemq-go/pkg/uuid"
 )
@@ -38,10 +39,15 @@ type broker struct {
 // implementation (`pubsub.topic`). This allows to keep the topic name
 // agnostic to their underlying implementation.
 //
+// You can instance a write-only broker by passing no SQS related configuration options.
+// This broker will only be able to publish messages to SNS topics, but will not be able to
+// subscribe handlers to topics nor receive messages from SQS queues. Of course, no goroutine
+// to poll the SQS queue will be started in this case.
+//
 // IMPORTANT: this broker must be used in conjunction with a Codec middleware in
 // order to ensure that the messages are properly encoded and decoded.
 // Otherwise, only binary messages will be accepted when publishing or
-// delivering messages.
+// delivering messages. Hence, `WithCodec` option is mandatory.
 func NewBroker(ctx context.Context, option ...Option) (pubsub.Broker, error) {
 	opts := &options{
 		deliverTimeout:       5 * time.Second,
@@ -59,12 +65,12 @@ func NewBroker(ctx context.Context, option ...Option) (pubsub.Broker, error) {
 		return nil, errors.New("no SNS client was provided")
 	}
 
-	if opts.sqsClient == nil {
-		return nil, errors.New("no SQS client was provided")
+	if err := opts.validateSQSSettings(); err != nil {
+		return nil, err
 	}
 
-	if opts.sqsQueueURL == "" {
-		return nil, errors.New("no SQS queue URL was provided")
+	if opts.codec == nil {
+		return nil, errors.New("no codec was provided")
 	}
 
 	ctx, cancel := context.WithCancel(ctx)
@@ -84,7 +90,7 @@ func NewBroker(ctx context.Context, option ...Option) (pubsub.Broker, error) {
 		go b.run()
 	}()
 
-	return b, nil
+	return codec.NewCodecMiddleware(b, opts.codec), nil
 }
 
 func (b *broker) Publish(ctx context.Context, topic pubsub.Topic, m interface{}) error {
@@ -111,6 +117,10 @@ func (b *broker) Publish(ctx context.Context, topic pubsub.Topic, m interface{})
 }
 
 func (b *broker) Subscribe(ctx context.Context, topic pubsub.Topic, handler pubsub.Handler, option ...pubsub.SubscribeOption) (pubsub.Subscription, error) {
+	if b.options.isWriteOnly() {
+		return nil, errors.New("no SQS queue ARN provided, broker is configured as write-only")
+	}
+
 	topicARN, err := b.topics.arnOf(topic)
 	if err != nil {
 		return nil, err
@@ -196,6 +206,10 @@ func (b *broker) Shutdown(ctx context.Context) error {
 }
 
 func (b *broker) run() {
+	if b.options.isWriteOnly() {
+		return
+	}
+
 	done := b.runnerCtx.Done()
 	topicTicker := time.NewTicker(b.options.topicsReloadInterval)
 	defer topicTicker.Stop()

provider/snssqs/broker_test.go

@@ -226,6 +226,41 @@ func TestMultiHostBroker(t *testing.T) {
 	})
 }
 
+func TestWriteOnlyBroker(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	t.Run("GIVEN a sns topic AND no sqs configurations", func(t *testing.T) {
+		cfg := awsConfig(t)
+		snsCli := awssns.NewFromConfig(cfg)
+
+		topic := pubsub.Topic("test-topic")
+		topicARN, err := prepareTopic(ctx, snsCli, "test-topic")
+		require.NoError(t, err)
+		require.NotEmpty(t, topicARN)
+
+		broker, err := snssqs.NewBroker(ctx,
+			snssqs.WithSNSClient(snsCli),
+			snssqs.WithCodec(codec.JSON),
+		)
+
+		t.Run("WHEN we instance the broker with just sns configuration", func(t *testing.T) {
+			t.Run("THEN there is no error", func(t *testing.T) {
+				require.NoError(t, err)
+			})
+
+			t.Run("AND we can publish messages to the topic", func(t *testing.T) {
+				require.NoError(t, broker.Publish(ctx, topic, "hello world"))
+			})
+
+			t.Run("BUT we cannot subscribe to the topic", func(t *testing.T) {
+				_, err := broker.Subscribe(ctx, topic, pubsub.NewHandler((&consumer{}).handle))
+				require.Error(t, err)
+			})
+		})
+	})
+}
+
 type consumer struct {
 	rcv queue
 	mu  sync.RWMutex
@@ -303,13 +338,14 @@ func prepareBroker(
 		snssqs.WithSQSQueueARN(attr.Attributes[string(sqstypes.QueueAttributeNameQueueArn)]),
 		snssqs.WithSQSQueueURL(aws.ToString(qRes.QueueUrl)),
 		snssqs.WithWaitTimeSeconds(1),
+		snssqs.WithCodec(codec.JSON),
 	)
 
 	if err != nil {
 		return nil, err
 	}
 
-	return codec.NewCodecMiddleware(broker, codec.JSON), nil
+	return broker, nil
 }
 
 func prepareTopic(ctx context.Context, cli *awssns.Client, topic pubsub.Topic) (string, error) {

provider/snssqs/options.go

@@ -1,7 +1,10 @@
 package snssqs
 
 import (
+	"errors"
 	"time"
+
+	"github.com/botchris/go-pubsub/middleware/codec"
 )
 
 // Option defines a function signature for configuration options.
@@ -19,6 +22,7 @@ type options struct {
 	maxMessages          int32
 	visibilityTimeout    int32
 	waitTimeSeconds      int32
+	codec                codec.Codec
 }
 
 type fnOption struct {
@@ -120,3 +124,38 @@ func WithWaitTimeSeconds(t int32) Option {
 		},
 	}
 }
+
+// WithCodec sets the codec to be used by the broker.
+func WithCodec(codec codec.Codec) Option {
+	return fnOption{
+		f: func(o *options) {
+			o.codec = codec
+		},
+	}
+}
+
+func (o options) validateSQSSettings() error {
+	if o.isWriteOnly() {
+		return nil
+	}
+
+	if o.sqsQueueURL == "" {
+		return errors.New("no SQS queue URL was provided")
+	}
+
+	if o.sqsQueueARN == "" {
+		return errors.New("no SQS queue ARN was provided")
+	}
+
+	if o.sqsClient == nil {
+		return errors.New("no SQS client was provided")
+	}
+
+	return nil
+}
+
+func (o options) isWriteOnly() bool {
+	return o.sqsQueueURL == "" &&
+		o.sqsQueueARN == "" &&
+		o.sqsClient == nil
+}