func (ss *SNSServer) listenAndPublishMessage() {
	for {
		select {
		case message := <-ss.notificationData:
			params := &sns.PublishInput{
				Message: aws.String(message),  
				MessageAttributes: map[string]*sns.MessageAttributeValue{
					MSG_ATTR: {  
						DataType:    aws.String("String"),  
						StringValue: aws.String(ss.Config.Env),
					},
				},
				Subject:  aws.String("new webhook"),
				TopicArn: aws.String(ss.Config.Sns.TopicArn),
			}
			resp, err := ss.SVC.Publish(params)
			if err != nil {
				ss.errorLog.Log(logging.MessageKey(), "SNS send message error", logging.ErrorKey(), err)
			}
			ss.debugLog.Log(logging.MessageKey(), "SNS send message", "response", resp)
		}
	}
}
listenAndPublishMessage go routine listens for data on notificationData channel
and publishes it to SNS
This go Routine is started when SNS Ready
package aws
import (
	"container/list"
	"errors"
	"fmt"
	"net/http"
	"strconv"
	"strings"
	"time"
	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/service/sns"
	"github.com/gorilla/mux"
	"github.com/xmidt-org/webpa-common/v2/logging"
	"github.com/xmidt-org/webpa-common/v2/xhttp"
)
const (
	MSG_ATTR           = "scytale.env"
	SNS_VALIDATION_ERR = "SNS signature validation error"
)
type MsgAttr struct {
	Type  string
	Value string
}
type SNSMessage struct {
	Type              string
	MessageId         string
	Token             string
	TopicArn          string
	Subject           string
	Message           string
	SubscribeURL      string
	Timestamp         string
	SignatureVersion  string
	Signature         string
	SigningCertURL    string
	UnsubscribeURL    string
	MessageAttributes map[string]MsgAttr
}
func (ss *SNSServer) SetSNSRoutes(urlPath string, r *mux.Router, handler http.Handler) {
	r.HandleFunc(urlPath, ss.SubscribeConfirmHandle).Methods("POST").Headers("x-amz-sns-message-type", "SubscriptionConfirmation")
	if handler != nil {
		ss.debugLog.Log(logging.MessageKey(), "handler not nil", "urlPath", urlPath)
		r.Handle(urlPath, handler).Methods("POST").Headers("x-amz-sns-message-type", "Notification")
	} else {
		ss.debugLog.Log(logging.MessageKey(), "handler nil", "urlPath", urlPath)
		r.HandleFunc(urlPath, func(rw http.ResponseWriter, req *http.Request) {
			ss.NotificationHandle(rw, req)
		}).Methods("POST").Headers("x-amz-sns-message-type", "Notification")
	}
}
func (ss *SNSServer) Subscribe() {
	params := &sns.SubscribeInput{
		Protocol: aws.String(ss.SelfUrl.Scheme),       
		TopicArn: aws.String(ss.Config.Sns.TopicArn),  
		Endpoint: aws.String(ss.SelfUrl.String()),
	}
	ss.debugLog.Log("subscribeParams", params)
	failCounter := ss.metrics.SNSSubscribeAttempt.With("code", "failure")
	okayCounter := ss.metrics.SNSSubscribeAttempt.With("code", "okay")
	failCounter.Add(0.0)
	okayCounter.Add(0.0)
	resp, err := ss.SVC.Subscribe(params)
	if err == nil {
		okayCounter.Add(1.0)
	} else {
		failCounter.Add(1.0)
		attemptNum := 1
		ss.errorLog.Log(logging.MessageKey(), "SNS subscribe error", "attempt", attemptNum, logging.ErrorKey(), err)
		attemptNum++
		if *params.TopicArn == "arn:aws:sns:us-east-1:1234:test-topic" ||
			*params.Endpoint == "http://host:10000/api/v2/aws/sns" {
			return
		}
		for {
			time.Sleep(time.Second * 5)
			resp, err = ss.SVC.Subscribe(params)
			if err != nil {
				failCounter.Add(1.0)
				ss.errorLog.Log(logging.MessageKey(), "SNS subscribe error", "attempt", attemptNum, logging.ErrorKey(), err)
			} else {
				okayCounter.Add(1.0)
				break
			}
			attemptNum++
		}
	}
	ss.debugLog.Log("subscribeResponse", resp, logging.MessageKey(), "SNS is not yet ready", "subscriptionArn", *resp.SubscriptionArn)
}
func (ss *SNSServer) SubscribeConfirmHandle(rw http.ResponseWriter, req *http.Request) {
	msg := new(SNSMessage)
	raw, err := DecodeJSONMessage(req, msg)
	if err != nil {
		ss.errorLog.Log(logging.MessageKey(), "SNS read req body error", logging.ErrorKey(), err)
		xhttp.WriteError(rw, http.StatusBadRequest, "request body error")
		return
	}
	valid, v_err := ss.Validate(msg)
	if !valid || v_err != nil {
		ss.errorLog.Log(logging.MessageKey(), "SNS signature validation error", logging.ErrorKey(), v_err)
		xhttp.WriteError(rw, http.StatusBadRequest, SNS_VALIDATION_ERR)
		return
	}
	if !strings.EqualFold(msg.TopicArn, ss.Config.Sns.TopicArn) {
		ss.errorLog.Log(
			logging.MessageKey(), "SNS subscription confirmation TopicArn mismatch",
			"received", msg.TopicArn,
			"expected", ss.Config.Sns.TopicArn)
		xhttp.WriteError(rw, http.StatusBadRequest, "TopicArn does not match")
		return
	}
	ss.debugLog.Log(
		logging.MessageKey(), "SNS confirmation payload",
		"raw", string(raw),
		"msg", msg,
	)
	params := &sns.ConfirmSubscriptionInput{
		Token:    aws.String(msg.Token),     
		TopicArn: aws.String(msg.TopicArn),  
	}
	resp, err := ss.SVC.ConfirmSubscription(params)
	if err != nil {
		ss.errorLog.Log(logging.MessageKey(), "SNS confirm error", logging.ErrorKey(), err)
		return
	}
	ss.debugLog.Log(logging.MessageKey(), "SNS confirm response", "response", resp)
	ss.subscriptionArn.Store(*resp.SubscriptionArn)
	ss.debugLog.Log(logging.MessageKey(), "SNS is ready", "subscriptionArn", *resp.SubscriptionArn)
	ss.metrics.SNSSubscribed.Add(1.0)
	go ss.listenAndPublishMessage()
	go ss.UnsubscribeOldSubscriptions()
}
func (ss *SNSServer) NotificationHandle(rw http.ResponseWriter, req *http.Request) []byte {
	subArn := req.Header.Get("X-Amz-Sns-Subscription-Arn")
	if !ss.ValidateSubscriptionArn(subArn) {
		xhttp.WriteError(rw, http.StatusInternalServerError, "SubscriptionARN does not match")
		ss.SNSNotificationReceivedCounter(http.StatusInternalServerError)
		return nil
	}
	msg := new(SNSMessage)
	raw, err := DecodeJSONMessage(req, msg)
	if err != nil {
		ss.errorLog.Log(logging.MessageKey(), "SNS read req body error", logging.ErrorKey(), err)
		xhttp.WriteError(rw, http.StatusBadRequest, "request body error")
		ss.SNSNotificationReceivedCounter(http.StatusBadRequest)
		return nil
	}
	valid, v_err := ss.Validate(msg)
	if !valid || v_err != nil {
		ss.errorLog.Log(logging.MessageKey(), "SNS signature validation error", logging.ErrorKey(), v_err)
		xhttp.WriteError(rw, http.StatusBadRequest, SNS_VALIDATION_ERR)
		ss.SNSNotificationReceivedCounter(http.StatusBadRequest)
		return nil
	}
	ss.debugLog.Log(
		logging.MessageKey(), "SNS notification payload",
		"raw", string(raw),
		"msg", msg,
	)
	if !strings.EqualFold(msg.TopicArn, ss.Config.Sns.TopicArn) {
		ss.errorLog.Log(
			logging.MessageKey(), "SNS notification TopicArn mismatch",
			"received", msg.TopicArn,
			"expected", ss.Config.Sns.TopicArn)
		xhttp.WriteError(rw, http.StatusBadRequest, "TopicArn does not match")
		ss.SNSNotificationReceivedCounter(http.StatusBadRequest)
		return nil
	}
	EnvAttr := msg.MessageAttributes[MSG_ATTR]
	msgEnv := EnvAttr.Value
	ss.debugLog.Log(logging.MessageKey(), "SNS notification", "envAttr", EnvAttr, "msgEnv", msgEnv)
	if msgEnv != ss.Config.Env {
		ss.errorLog.Log(logging.MessageKey(), "SNS environment mismatch", "msgEnv", msgEnv, "config", ss.Config.Env)
		xhttp.WriteError(rw, http.StatusBadRequest, "SNS Msg config env does not match")
		ss.SNSNotificationReceivedCounter(http.StatusBadRequest)
		return nil
	}
	ss.SNSNotificationReceivedCounter(http.StatusOK)
	return []byte(msg.Message)
}
func (ss *SNSServer) PublishMessage(message string) error {
	ss.debugLog.Log(logging.MessageKey(), "SNS PublishMessage", "called", message)
	ss.metrics.SNSNotificationSent.Add(1.0)
	timer := time.NewTimer(ss.channelClientTimeout)
	defer timer.Stop()
	select {
	case ss.notificationData <- message:
		return nil
	case <-timer.C:
		return errors.New("Unable to add message to channel in allotted time.")
	}
}