pubsub_push_handler.go

package main

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

// Google pubsub message envelope
type PubsubMessage struct {
	Message Message `json:"message"`
}

// Google pubsub message
type Message struct {
	PublishTime string            `json:"publishTime"`
	Attributes  map[string]string `json:"attributes"`
	Data        string            `json:"data"`
}

// Optional payload structure (if we want to send data beyond the attributes)
type MyEventPayload struct {
	MyField string `json:"my_field"`
}

func main() {

	handler := func(w http.ResponseWriter, req *http.Request) {

		var pubsubMessage PubsubMessage
		decoderBody := json.NewDecoder(req.Body)
		if err := decoderBody.Decode(&pubsubMessage); err != nil {
			log.Printf("unmarshal pubsub push message: %v\n", err)
			w.WriteHeader(http.StatusBadRequest)
			return
		}

		// Debug
		// fmt.Printf("PublishTime: %s\n", pubsubMessage.Message.PublishTime)
		// fmt.Printf("Data: %s\n", pubsubMessage.Message.Data)
		// for k, v := range pubsubMessage.Message.Attributes {
		//   fmt.Printf("  %s => %s\n", k, v)
		// }

		fmt.Printf("Received change event from eventlog:\n")
		fmt.Printf("  timestamp:    %s\n", pubsubMessage.Message.Attributes["timestamp"])
		fmt.Printf("  environment:  %s\n", pubsubMessage.Message.Attributes["environment"])
		fmt.Printf("  rsid:         %s\n", pubsubMessage.Message.Attributes["rsid"])
		fmt.Printf("  resource_uri: %s\n", pubsubMessage.Message.Attributes["resource_uri"])
		fmt.Printf("  change_type:  %s\n", pubsubMessage.Message.Attributes["change_type"])

		// This is all we actually need - you can now get the data using HTTP GET on the
		// provided resource URI.

		// Optionally do something with the message data
		if pubsubMessage.Message.Data != "" {
			jsonBytes, err := base64.StdEncoding.DecodeString(pubsubMessage.Message.Data)
			if err != nil {
				log.Printf("base64 decode message data: %v\n", err)
				w.WriteHeader(http.StatusBadRequest)
				return
			}

			var payload MyEventPayload
			if err := json.Unmarshal(jsonBytes, &payload); err != nil {
				log.Printf("unmarshall event payload: %v\n", err)
				w.WriteHeader(http.StatusBadRequest)
				return
			}
		}

		w.WriteHeader(http.StatusNoContent)
	}

	http.HandleFunc("/events", handler)

	err := http.ListenAndServe(":8080", nil)
	if err != nil {
		log.Fatal(err)
	}

}