Milo: Buildbucket PubSub ingestion outline

milo/buildsource/buildbot/pubsub.go

Index: milo/buildsource/buildbot/pubsub.go
diff --git a/milo/buildsource/buildbot/pubsub.go b/milo/buildsource/buildbot/pubsub.go
index 15814813e7e37d706a503cf05c35f82e5b82eff0..83870bf7ef93f21aec2ef6ae9f6ce02fc4da4073 100644
--- a/milo/buildsource/buildbot/pubsub.go
+++ b/milo/buildsource/buildbot/pubsub.go
@@ -8,7 +8,6 @@ import (
 	"bytes"
 	"compress/gzip"
 	"compress/zlib"
-	"encoding/base64"
 	"encoding/json"
 	"fmt"
 	"net/http"
@@ -51,17 +50,6 @@ var (
 		field.String("status"))
 )
 
-type pubSubMessage struct {
-	Attributes map[string]string `json:"attributes"`
-	Data       string            `json:"data"`
-	MessageID  string            `json:"message_id"`
-}
-
-type pubSubSubscription struct {
-	Message      pubSubMessage `json:"message"`
-	Subscription string        `json:"subscription"`
-}
-
 type buildMasterMsg struct {
 	Master *buildbotMaster  `json:"master"`
 	Builds []*buildbotBuild `json:"builds"`
@@ -112,11 +100,6 @@ func putDSMasterJSON(
 	return ds.Put(c, &entry)
 }
 
-// GetData returns the expanded form of Data (decoded from base64).
-func (m *pubSubSubscription) GetData() ([]byte, error) {
-	return base64.StdEncoding.DecodeString(m.Message.Data)
-}
-
 // unmarshal a gzipped byte stream into a list of buildbot builds and masters.
 func unmarshal(
 	c context.Context, msg []byte) ([]*buildbotBuild, *buildbotMaster, error) {
@@ -293,7 +276,7 @@ func PubSubHandler(ctx *router.Context) {
 // a status code.  StatusOK (200) for okay (ACK implied, don't retry).
 // Anything else will signal to pubsub to retry.
 func pubSubHandlerImpl(c context.Context, r *http.Request) int {
-	msg := pubSubSubscription{}
+	msg := common.PubSubSubscription{}
 	now := int(clock.Now(c).Unix())
 	defer r.Body.Close()
 	dec := json.NewDecoder(r.Body)