Server that consumes HTTP messages sent from a SNS subscription.

sqs-consumer.go

package main

import (
	"encoding/json"
	"fmt"
	"log"
	"net/http"
)

const (
	headerAmzSnsMsgType = "x-amz-sns-message-type"
)

// SnsMsg is a generic base for topic messages.
type SnsMsg struct {
	Type             string `json:"Type"`
	MessageId        string `json:"MessageId"`
	TopicArn         string `json:"TopicArn"`
	Message          string `json:"Message"`
	Timestamp        string `json:"Timestamp"`
	SignatureVersion string `json:"Version"`
	Signature        string `json:"Signature"`
	SigningCertURL   string `json:SigningCertURL"`
}

// verify the signature of the message.
func (m SnsMsg) verify() error {
	// TODO: implement this
	return nil
}

// SubscriptionConfirmationMsg contain information for confirming a topic subscription.
type SubscriptionConfirmationMsg struct {
	SnsMsg
	Token        string `json:"Token"`
	SubscribeURL string `json:"SubscribeURL"`
}

// NotificationMsg contain generic plaintext messages from a topic.
type NotificationMsg struct {
	SnsMsg
	UnsubscribeURL string `json:"UnsubscribeURL"`
	Subject        string `json:"Subject"`
}
​
// notification handles all message types.
func notification(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodPost {
		http.NotFound(w, r)
		return
	}

	switch msgType := r.Header.Get(headerAmzSnsMsgType); msgType {
	case "SubscriptionConfirmation":
		handleSubscriptionConfirmation(w, r)
	case "Notification":
		handleNotification(w, r)
	default:
		errMsg := fmt.Sprintf("Header '%s' must be one of 'SubscriptionConfirmation' or 'Notification'", headerAmzSnsMsgType)
		http.Error(w, errMsg, http.StatusBadRequest)
	}
}

// handleSubscriptionConfirmation responds to messages of type 'SubscriptionConfirmation'.
func handleSubscriptionConfirmation(w http.ResponseWriter, r *http.Request) {
	decoder := json.NewDecoder(r.Body)
	var msg SubscriptionConfirmationMsg
	if err := decoder.Decode(&msg); err != nil {
		errMsg := fmt.Sprintf("The message is malformed: %s", err.Error())
		http.Error(w, errMsg, http.StatusBadRequest)
		return
	}

	if err := msg.verify(); err != nil {
		errMsg := fmt.Sprintf("Unable to verify message: %s", err.Error())
		http.Error(w, errMsg, http.StatusUnauthorized)
		return
	}

	resp, err := http.Get(msg.SubscribeURL)
	if err != nil || resp.StatusCode != http.StatusOK {
		if err == nil {
			err = fmt.Errorf("subscription url responded with '%s'", resp.Status)
		}
		errMsg := fmt.Sprintf("There was an error requesting the subscription url '%s': %s", msg.SubscribeURL, err.Error())
		http.Error(w, errMsg, http.StatusInternalServerError)
	}
}

// handleNotification responds and processes messages of type 'Notification'.
func handleNotification(w http.ResponseWriter, r *http.Request) {
	decoder := json.NewDecoder(r.Body)
	var msg NotificationMsg
	if err := decoder.Decode(&msg); err != nil {
		errMsg := fmt.Sprintf("The message is malformed: %s", err.Error())
		http.Error(w, errMsg, http.StatusBadRequest)
		return
	}

	if err := msg.verify(); err != nil {
		errMsg := fmt.Sprintf("Unable to verify message: %s", err.Error())
		http.Error(w, errMsg, http.StatusUnauthorized)
		return
	}

	log.Printf("%+v\n", msg)
}

func main() {
	http.HandleFunc("/notification", notification)
	log.Println("Serving notification client at 127.0.0.1:8080")
	log.Fatal(http.ListenAndServe(":8080", nil))
}