pubsub/aws/snssqs to support SQS dead-letter queue

pubsub/aws/snssqs/snssqs.go

@@ -2,12 +2,12 @@ package snssqs
 
 import (
 	"context"
-	"crypto/sha256"
 	"encoding/json"
 	"errors"
 	"fmt"
 	"strconv"
 	"strings"
+	"regexp"
 
 	"github.com/aws/aws-sdk-go/aws"
 	sns "github.com/aws/aws-sdk-go/service/sns"
@@ -20,8 +20,8 @@ import (
 type snsSqs struct {
 	// key is the topic name, value is the ARN of the topic
 	topics map[string]string
-	// key is the hashed topic name, value is the actual topic name
-	topicHash map[string]string
+	// key is the sanitized topic name, value is the actual topic name
+	topicSanitized map[string]string
 	// key is the topic name, value holds the ARN of the queue and its url
 	queues        map[string]*sqsQueueInfo
 	snsClient     *sns.SNS
@@ -66,6 +66,9 @@ const (
 	awsSnsTopicNameKey = "dapr-topic-name"
 )
 
+var awsSnsSqsAllowedCharsRe = regexp.MustCompile("[^a-zA-Z0-9_\\-]+")
+
+
 func NewSnsSqs(l logger.Logger) pubsub.PubSub {
 	return &snsSqs{
 		logger:        l,
@@ -93,13 +96,16 @@ func parseInt64(input string, propertyName string) (int64, error) {
 	return int64(number), nil
 }
 
-// take a name and hash it for compatibility with AWS resource names
-// the output is fixed at 64 characters
-func nameToHash(name string) string {
-	h := sha256.New()
-	h.Write([]byte(name))
 
-	return fmt.Sprintf("%x", h.Sum(nil))
+// sanitize topic/queue name to conform with:
+// https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/quotas-queues.html
+func nameToAWSSanitizedName(name string) string {
+	sanitizedName := awsSnsSqsAllowedCharsRe.ReplaceAllString(name, "")
+	if len(sanitizedName) > 80 {
+		sanitizedName = sanitizedName[:80]
+	}
+
+	return sanitizedName
 }
 
 func (s *snsSqs) getSnsSqsMetatdata(metadata pubsub.Metadata) (*snsSqsMetadata, error) {
@@ -207,7 +213,7 @@ func (s *snsSqs) Init(metadata pubsub.Metadata) error {
 	// both Publish and Subscribe need reference the topic ARN
 	// track these ARNs in this map
 	s.topics = make(map[string]string)
-	s.topicHash = make(map[string]string)
+	s.topicSanitized = make(map[string]string)
 	s.queues = make(map[string]*sqsQueueInfo)
 	sess, err := aws_auth.GetClient(md.AccessKey, md.SecretKey, md.SessionToken, md.Region, md.Endpoint)
 	if err != nil {
@@ -220,16 +226,16 @@ func (s *snsSqs) Init(metadata pubsub.Metadata) error {
 }
 
 func (s *snsSqs) createTopic(topic string) (string, string, error) {
-	hashedName := nameToHash(topic)
+	sanitizedName := nameToAWSSanitizedName(topic)
 	createTopicResponse, err := s.snsClient.CreateTopic(&sns.CreateTopicInput{
-		Name: aws.String(hashedName),
+		Name: aws.String(sanitizedName),
 		Tags: []*sns.Tag{{Key: aws.String(awsSnsTopicNameKey), Value: aws.String(topic)}},
 	})
 	if err != nil {
 		return "", "", err
 	}
 
-	return *(createTopicResponse.TopicArn), hashedName, nil
+	return *(createTopicResponse.TopicArn), sanitizedName, nil
 }
 
 // get the topic ARN from the topics map. If it doesn't exist in the map, try to fetch it from AWS, if it doesn't exist
@@ -245,7 +251,7 @@ func (s *snsSqs) getOrCreateTopic(topic string) (string, error) {
 
 	s.logger.Debugf("No topic ARN found for %s\n Creating topic instead.", topic)
 
-	topicArn, hashedName, err := s.createTopic(topic)
+	topicArn, sanitizedName, err := s.createTopic(topic)
 	if err != nil {
 		s.logger.Errorf("error creating new topic %s: %v", topic, err)
 
@@ -254,14 +260,14 @@ func (s *snsSqs) getOrCreateTopic(topic string) (string, error) {
 
 	// record topic ARN
 	s.topics[topic] = topicArn
-	s.topicHash[hashedName] = topic
+	s.topicSanitized[sanitizedName] = topic
 
 	return topicArn, nil
 }
 
 func (s *snsSqs) createQueue(queueName string) (*sqsQueueInfo, error) {
 	createQueueResponse, err := s.sqsClient.CreateQueue(&sqs.CreateQueueInput{
-		QueueName: aws.String(nameToHash(queueName)),
+		QueueName: aws.String(nameToAWSSanitizedName(queueName)),
 		Tags:      map[string]*string{awsSqsQueueNameKey: aws.String(queueName)},
 	})
 	if err != nil {
@@ -397,7 +403,7 @@ func (s *snsSqs) handleMessage(message *sqs.Message, queueInfo *sqsQueueInfo, ha
 	}
 
 	topic := parseTopicArn(messageBody.TopicArn)
-	topic = s.topicHash[topic]
+	topic = s.topicSanitized[topic]
 	err = handler(context.Background(), &pubsub.NewMessage{
 		Data:  []byte(messageBody.Message),
 		Topic: topic,
@@ -491,11 +497,12 @@ func (s *snsSqs) Subscribe(req pubsub.SubscribeRequest, handler pubsub.Handler)
 }
 
 func (s *snsSqs) Close() error {
-	for _, sub := range s.subscriptions {
-		s.snsClient.Unsubscribe(&sns.UnsubscribeInput{
-			SubscriptionArn: sub,
-		})
-	}
+	s.logger.Debugf("Close was called and is now NOOP")
+	// for _, sub := range s.subscriptions {
+	// 	s.snsClient.Unsubscribe(&sns.UnsubscribeInput{
+	// 		SubscriptionArn: sub,
+	// 	})
+	// }
 
 	return nil
 }

pubsub/aws/snssqs/snssqs_test.go

@@ -1,8 +1,6 @@
 package snssqs
 
 import (
-	"fmt"
-	"strings"
 	"testing"
 
 	"github.com/dapr/components-contrib/pubsub"
@@ -236,22 +234,14 @@ func Test_parseInt64(t *testing.T) {
 	r.Error(err)
 }
 
-func Test_nameToHash(t *testing.T) {
+
+func Test_replaceNameToAWSSanitizedName(t *testing.T) {
 	r := require.New(t)
 
-	// This string is too long and contains invalid character for either an SQS queue or an SNS topic
-	hashedName := nameToHash(`
-		Some invalid name // for an AWS resource &*()*&&^Some invalid name // for an AWS resource &*()*&&^Some invalid 
+	s := `Some_invalid-name // for an AWS resource &*()*&&^Some invalid name // for an AWS resource &*()*&&^Some invalid 
 		name // for an AWS resource &*()*&&^Some invalid name // for an AWS resource &*()*&&^Some invalid name // for an
-		AWS resource &*()*&&^Some invalid name // for an AWS resource &*()*&&^
-	`)
-
-	r.Equal(64, len(hashedName))
-	// Output is only expected to contain lower case characters representing valid hexadecimal numerals
-	for _, c := range hashedName {
-		r.True(
-			strings.ContainsAny(
-				"abcdef0123456789", string(c)),
-			fmt.Sprintf("Invalid character %s in hashed name", string(c)))
-	}
+		AWS resource &*()*&&^Some invalid name // for an AWS resource &*()*&&^`
+	v := nameToAWSSanitizedName(s)
+	r.Equal(80, len(v))
+	r.Equal("Some_invalid-nameforanAWSresourceSomeinvalidnameforanAWSresourceSomeinvalidnamef", v)
 }