Sns publish batch

.github/README.md

@@ -43,6 +43,7 @@ Here is a list of the APIs:
  - [x] Subscribe (raw)
  - [x] ListSubscriptions
  - [x] Publish
+ - [x] PublishBatch
  - [x] DeleteTopic
  - [x] Subscribe
  - [x] Unsubscribe

app/gosns/publish.go

@@ -38,27 +38,27 @@ func PublishV1(req *http.Request) (int, interfaces.AbstractResponseBody) {
 	topicName := arnSegments[len(arnSegments)-1]
 
 	_, ok = app.SyncTopics.Topics[topicName]
-	if ok {
-		log.WithFields(log.Fields{
-			"topic":    topicName,
-			"topicArn": requestBody.TopicArn,
-			"subject":  requestBody.Subject,
-		}).Debug("Publish to Topic")
-		for _, subscription := range app.SyncTopics.Topics[topicName].Subscriptions {
-			switch app.Protocol(subscription.Protocol) {
-			case app.ProtocolSQS:
-				err := publishSQS(subscription, topicName, requestBody)
-				if err != nil {
-					utils.CreateErrorResponseV1(err.Error(), false)
-				}
-			case app.ProtocolHTTP:
-				fallthrough
-			case app.ProtocolHTTPS:
-				publishHTTP(subscription, requestBody)
+	if !ok {
+		return utils.CreateErrorResponseV1("TopicNotFound", false)
+	}
+
+	log.WithFields(log.Fields{
+		"topic":    topicName,
+		"topicArn": requestBody.TopicArn,
+		"subject":  requestBody.Subject,
+	}).Debug("Publish to Topic")
+	for _, subscription := range app.SyncTopics.Topics[topicName].Subscriptions {
+		switch app.Protocol(subscription.Protocol) {
+		case app.ProtocolSQS:
+			err := publishSQS(subscription, topicName, requestBody)
+			if err != nil {
+				utils.CreateErrorResponseV1(err.Error(), false)
 			}
+		case app.ProtocolHTTP:
+			fallthrough
+		case app.ProtocolHTTPS:
+			publishHTTP(subscription, requestBody.TopicArn, requestBody)
 		}
-	} else {
-		return utils.CreateErrorResponseV1("TopicNotFound", false)
 	}
 
 	//Create the response
@@ -74,9 +74,15 @@ func PublishV1(req *http.Request) (int, interfaces.AbstractResponseBody) {
 	return http.StatusOK, respStruct
 }
 
-func publishSQS(subscription *app.Subscription, topicName string, requestBody *models.PublishRequest) error {
-	messageAttributes := utils.ConvertToOldMessageAttributeValueStructure(requestBody.MessageAttributes)
-	if subscription.FilterPolicy != nil && !subscription.FilterPolicy.IsSatisfiedBy(messageAttributes) {
+type MessageCreator interface {
+	GetMessageAttributes() map[string]app.MessageAttributeValue
+	GetMessage() string
+	GetSubject() string
+	GetMessageStructure() string
+}
+
+func publishSQS(subscription *app.Subscription, topicName string, messager MessageCreator) error {
+	if subscription.FilterPolicy != nil && !subscription.FilterPolicy.IsSatisfiedBy(messager.GetMessageAttributes()) {
 		return nil
 	}
 
@@ -86,49 +92,47 @@ func publishSQS(subscription *app.Subscription, topicName string, requestBody *m
 	arnSegments := strings.Split(queueName, ":")
 	queueName = arnSegments[len(arnSegments)-1]
 
-	if _, ok := app.SyncQueues.Queues[queueName]; ok {
-		msg := app.Message{}
-
-		if subscription.Raw == false {
-			m, err := createMessageBody(subscription, requestBody.Message, requestBody.Subject, requestBody.MessageStructure, messageAttributes)
-			if err != nil {
-				return err
-			}
-
-			msg.MessageBody = m
+	if _, ok := app.SyncQueues.Queues[queueName]; !ok {
+		log.Infof("Queue %s does not exist, message discarded\n", queueName)
+		return nil
+	}
+	msg := app.Message{}
+	if subscription.Raw == false {
+		m, err := createMessageBody(subscription, messager.GetMessage(), messager.GetSubject(), messager.GetMessageStructure(), messager.GetMessageAttributes())
+		if err != nil {
+			return err
+		}
+		msg.MessageBody = m
+	} else {
+		msg.MessageAttributes = messager.GetMessageAttributes()
+		msg.MD5OfMessageAttributes = common.HashAttributes(msg.MessageAttributes)
+		m, err := extractMessageFromJSON(messager.GetMessage(), subscription.Protocol)
+		if err == nil {
+			msg.MessageBody = []byte(m)
 		} else {
-			msg.MessageAttributes = messageAttributes
-			msg.MD5OfMessageAttributes = common.HashAttributes(messageAttributes)
-			m, err := extractMessageFromJSON(requestBody.Message, subscription.Protocol)
-			if err == nil {
-				msg.MessageBody = []byte(m)
-			} else {
-				msg.MessageBody = []byte(requestBody.Message)
-			}
+			msg.MessageBody = []byte(messager.GetMessage())
 		}
+	}
+	msg.MD5OfMessageBody = common.GetMD5Hash(messager.GetMessage())
+	msg.Uuid = uuid.NewString()
 
-		msg.MD5OfMessageBody = common.GetMD5Hash(requestBody.Message)
-		msg.Uuid, _ = common.NewUUID()
-		app.SyncQueues.Lock()
-		app.SyncQueues.Queues[queueName].Messages = append(app.SyncQueues.Queues[queueName].Messages, msg)
-		app.SyncQueues.Unlock()
+	app.SyncQueues.Lock()
+	app.SyncQueues.Queues[queueName].Messages = append(app.SyncQueues.Queues[queueName].Messages, msg)
+	app.SyncQueues.Unlock()
 
-		log.Infof("%s: Topic: %s(%s), Message: %s\n", time.Now().Format("2006-01-02 15:04:05"), topicName, queueName, msg.MessageBody)
-	} else {
-		log.Infof("%s: Queue %s does not exist, message discarded\n", time.Now().Format("2006-01-02 15:04:05"), queueName)
-	}
+	log.Infof("Topic: %s(%s), Message: %s\n", topicName, queueName, msg.MessageBody)
 	return nil
 }
 
-func publishHTTP(subs *app.Subscription, requestBody *models.PublishRequest) {
-	messageAttributes := utils.ConvertToOldMessageAttributeValueStructure(requestBody.MessageAttributes)
+func publishHTTP(subs *app.Subscription, topicArn string, messager MessageCreator) {
+	messageAttributes := messager.GetMessageAttributes()
 	id := uuid.NewString()
 	msg := app.SNSMessage{
 		Type:              "Notification",
 		MessageId:         id,
-		TopicArn:          requestBody.TopicArn,
-		Subject:           requestBody.Subject,
-		Message:           requestBody.Message,
+		Message:           messager.GetMessage(),
+		TopicArn:          topicArn,
+		Subject:           messager.GetSubject(),
 		Timestamp:         time.Now().UTC().Format(time.RFC3339),
 		SignatureVersion:  "1",
 		SigningCertURL:    fmt.Sprintf("http://%s:%s/SimpleNotificationService/%s.pem", app.CurrentEnvironment.Host, app.CurrentEnvironment.Port, id),

app/gosns/publish_batch.go

@@ -0,0 +1,106 @@
+package gosns
+
+import (
+	"net/http"
+	"strings"
+
+	"github.com/Admiral-Piett/goaws/app"
+	"github.com/Admiral-Piett/goaws/app/interfaces"
+	"github.com/Admiral-Piett/goaws/app/models"
+	"github.com/Admiral-Piett/goaws/app/utils"
+	"github.com/google/uuid"
+	log "github.com/sirupsen/logrus"
+)
+
+func PublishBatchV1(req *http.Request) (int, interfaces.AbstractResponseBody) {
+	requestBody := models.NewPublishBatchRequest()
+	ok := utils.REQUEST_TRANSFORMER(requestBody, req, false)
+	if !ok {
+		log.Error("Invalid Request - PublishBatchV1")
+		return utils.CreateErrorResponseV1("InvalidParameterValue", false)
+	}
+
+	if requestBody.TopicArn == "" {
+		return utils.CreateErrorResponseV1("InvalidParameterValue", false)
+	}
+
+	if len(requestBody.PublishBatchRequestEntries.Member) == 0 {
+		return utils.CreateErrorResponseV1("EmptyBatchRequest", false)
+	}
+	if len(requestBody.PublishBatchRequestEntries.Member) > 10 {
+		return utils.CreateErrorResponseV1("TooManyEntriesInBatchRequest", false)
+	}
+
+	topicArn := requestBody.TopicArn
+	arnSegments := strings.Split(topicArn, ":")
+	topicName := arnSegments[len(arnSegments)-1]
+	topic, ok := app.SyncTopics.Topics[topicName]
+	if !ok {
+		return utils.CreateErrorResponseV1("TopicNotFound", false)
+	}
+
+	seen := make(map[string]bool)
+
+	for _, entry := range requestBody.PublishBatchRequestEntries.Member {
+		if entry == nil { // we use gorilla schema to parse value params. Indexing on the aws client starts at 1 but gorilla schema starts at 0 so we may have a nil entry at the start of the slice
+			continue
+		}
+		if entry.ID == "" {
+			// This is a required field for the PublishBatchRequestEntry entity but doesn't seem required in the request.
+			// If it's not present in the request then assume we should generate one.
+			entry.ID = uuid.NewString()
+		}
+		if seen[entry.ID] {
+			return utils.CreateErrorResponseV1("BatchEntryIdsNotDistinct", false)
+		}
+		seen[entry.ID] = true
+	}
+
+	successfulEntries := []models.PublishBatchResultEntry{}
+	failedEntries := []models.BatchResultErrorEntry{}
+	for _, entry := range requestBody.PublishBatchRequestEntries.Member {
+		if entry == nil { // we use gorilla schema to parse value params. Indexing on the aws client starts at 1 but gorilla schema starts at 0 so we may have a nil entry at the start of the slice
+			continue
+		}
+		// we now know all the entry.IDs are unique and non-blank
+		for _, sub := range topic.Subscriptions {
+			switch app.Protocol(sub.Protocol) {
+			case app.ProtocolSQS:
+				if err := publishSQS(sub, topicName, entry); err != nil {
+					er := models.SnsErrors[err.Error()]
+					failedEntries = append(failedEntries, models.BatchResultErrorEntry{
+						Code:        er.Code,
+						Id:          entry.ID,
+						Message:     er.Message,
+						SenderFault: true,
+					})
+				} else {
+					msgId := uuid.NewString()
+					successfulEntries = append(successfulEntries, models.PublishBatchResultEntry{
+						Id:        entry.ID,
+						MessageId: msgId,
+					})
+				}
+			case app.ProtocolHTTP:
+				fallthrough
+			case app.ProtocolHTTPS:
+				publishHTTP(sub, topicArn, entry)
+				msgId := uuid.NewString()
+				successfulEntries = append(successfulEntries, models.PublishBatchResultEntry{
+					Id:        entry.ID,
+					MessageId: msgId,
+				})
+			}
+		}
+	}
+
+	respStruct := models.PublishBatchResponse{
+		Xmlns: models.BASE_XMLNS,
+		Result: models.PublishBatchResult{
+			Successful: models.PublishBatchSuccessful{SuccessEntries: successfulEntries},
+			Failed:     models.PublishBatchFailed{ErrorEntries: failedEntries},
+		},
+		Metadata: app.ResponseMetadata{RequestId: uuid.NewString()},
+	}
+	return http.StatusOK, respStruct
+}