FIXED opts not being passed into SDK (#11)

* FIXED opts not being passed into SDK * Update pkg/queue/sqs.go Co-authored-by: Ricki Hastings <[email protected]> * Update pkg/pubsub/sns.go Co-authored-by: Ricki Hastings <[email protected]> * FIXED checking Co-authored-by: Ewan Valentine <[email protected]> Co-authored-by: Ricki Hastings <[email protected]>
peak-ai · Jun 17, 2020 · d6ae248 · d6ae248
1 parent d8940e6
commit d6ae248
Show file tree

Hide file tree

Showing 3 changed files with 48 additions and 2 deletions.
diff --git a/examples/main.go b/examples/main.go
@@ -8,7 +8,7 @@ import (
 )
 
 func main() {
-	discovery := d.NewDiscovery()
+	discovery := d.NewDiscovery(d.WithAWSBackend())
 	res, err := discovery.Request("acme-prod.scheduler->create-job", types.Request{
 		Body: []byte(`{ "frequency": "* * * * *", "type": "test" }`),
 	})

diff --git a/pkg/pubsub/sns.go b/pkg/pubsub/sns.go
@@ -19,17 +19,35 @@ func NewSNSAdapter(client *sns.SNS) *SNSAdapter {
 	return &SNSAdapter{client}
 }
 
+func (sa *SNSAdapter) parseOpts(opts types.Options) map[string]*sns.MessageAttributeValue {
+	atts := make(map[string]*sns.MessageAttributeValue, 0)
+	for key, val := range opts {
+		attributeValue, ok := val.(*sns.MessageAttributeValue)
+		if ok {
+			atts[key] = attributeValue
+		}
+	}
+
+	return atts
+}
+
 // Publish publishes an event to a queue
 func (sa *SNSAdapter) Publish(service *types.Service, request types.Request) error {
 	return sa.PublishWithOpts(service, request, types.Options{})
 }
 
-// PublishWithOpts -
+// PublishWithOpts takes the generic options type, converts to 'MessageAttributes'
 func (sa *SNSAdapter) PublishWithOpts(service *types.Service, request types.Request, opts types.Options) error {
 	input := &sns.PublishInput{
 		Message:  aws.String(string(request.Body)),
 		TopicArn: aws.String(service.Addr),
 	}
+
+	if len(opts) > 0 {
+		atts := sa.parseOpts(opts)
+		input.SetMessageAttributes(atts)
+	}
+
 	_, err := sa.client.Publish(input)
 	return err
 }

diff --git a/pkg/queue/sqs.go b/pkg/queue/sqs.go
@@ -22,12 +22,30 @@ func (qa *SQSAdapter) Queue(service *types.Service, request types.Request) (stri
 	return qa.QueueWithOpts(service, request, types.Options{})
 }
 
+func (qa *SQSAdapter) parseOpts(opts types.Options) map[string]*sqs.MessageAttributeValue {
+	atts := make(map[string]*sqs.MessageAttributeValue, 0)
+	for key, val := range opts {
+		attributeValue, ok := val.(*sqs.MessageAttributeValue)
+		if ok {
+			atts[key] = attributeValue
+		}
+	}
+
+	return atts
+}
+
 // QueueWithOpts queues a message, with options.
 func (qa *SQSAdapter) QueueWithOpts(service *types.Service, request types.Request, opts types.Options) (string, error) {
 	input := &sqs.SendMessageInput{
 		MessageBody: aws.String(string(request.Body)),
 		QueueUrl:    aws.String(service.Addr),
 	}
+
+	if len(opts) > 0 {
+		atts := qa.parseOpts(opts)
+		input.SetMessageAttributes(atts)
+	}
+
 	output, err := qa.client.SendMessage(input)
 	return *output.MessageId, err
 }
@@ -43,6 +61,16 @@ func (qa *SQSAdapter) ListenWithOpts(service *types.Service, opts types.Options)
 	input := &sqs.ReceiveMessageInput{
 		QueueUrl: aws.String(service.Addr),
 	}
+
+	// Options here are for keys only, so the format doesn't quite work here...
+	if len(opts) > 0 {
+		var keys []*string
+		for key, _ := range opts {
+			keys = append(keys, aws.String(key))
+		}
+		input.SetMessageAttributeNames(keys)
+	}
+
 	go func() {
 		for {
 			res, err := qa.client.ReceiveMessage(input)