Use native Pub/Sub library primitives.

appengine/tsmon/middleware.go

 // Copyright 2016 The Chromium Authors. All rights reserved.
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
 package tsmon
 
 import (
         "fmt"
         "net/http"
         "strings"
         "sync"
         "time"
 
         "github.com/golang/protobuf/proto"
         "github.com/julienschmidt/httprouter"
         "github.com/luci/gae/service/datastore"
         "github.com/luci/gae/service/info"
         gaeauth "github.com/luci/luci-go/appengine/gaeauth/client"
         "github.com/luci/luci-go/common/clock"
-        "github.com/luci/luci-go/common/gcloud/pubsub"
+        gcps "github.com/luci/luci-go/common/gcloud/pubsub"
         "github.com/luci/luci-go/common/logging"
         "github.com/luci/luci-go/common/tsmon"
         "github.com/luci/luci-go/common/tsmon/monitor"
         "github.com/luci/luci-go/common/tsmon/store"
         "github.com/luci/luci-go/common/tsmon/target"
         "github.com/luci/luci-go/server/middleware"
         "golang.org/x/net/context"
 )
 
 var (
@@ skipping 17 matching lines @@
                 flushIfNeeded(c)
         }
 }
 
 func initialize(c context.Context) error {
         var mon monitor.Monitor
         i := info.Get(c)
         if i.IsDevAppServer() {
                 mon = monitor.NewDebugMonitor("")
         } else {
-                client := func(c context.Context) (*http.Client, error) {
-                        // Create an HTTP client with the default appengine service account.
-                        auth, err := gaeauth.Authenticator(c, pubsub.PublisherScopes, nil)
-                        if err != nil {
-                                return nil, err
-                        }
-                        return auth.Client()
-                }
-
-                var err error
-                mon, err = monitor.NewPubsubMonitor(client, pubsubProject, pubsubTopic)
+                // Create an HTTP client with the default appengine service account.
+                auth, err := gaeauth.Authenticator(c, gcps.PublisherScopes, nil)
                 if err != nil {
                         return err
                 }
+                client, err := auth.Client()
+                if err != nil {
+                        return err
+                }
+
+                mon, err = monitor.NewPubsubMonitor(c, client, gcps.NewTopic(pubsubProject, pubsubTopic))
+                if err != nil {
+                        return err
+                }
         }
 
         // Create the target.
         tar := &target.Task{
                 DataCenter:  proto.String(targetDataCenter),
                 ServiceName: proto.String(i.AppID()),
                 JobName:     proto.String(i.ModuleName()),
                 HostName:    proto.String(strings.SplitN(i.VersionID(), ".", 2)[0]),
                 TaskNum:     proto.Int32(-1),
         }
@@ skipping 70 matching lines @@
                 if err := datastore.Get(c).Put(entity); err != nil {
                         logger.Errorf("Failed to update instance entity: %s", err)
                 }
         }()
 
         ret := tsmon.Flush(c)
 
         <-putDone
         return ret
 }