Use native Pub/Sub library primitives.

client/cmd/logdog_butler/output_pubsub.go

 // Copyright 2015 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 main
 
 import (
         "fmt"
         "time"
 
         "github.com/luci/luci-go/client/internal/logdog/butler/output"
-        "github.com/luci/luci-go/client/internal/logdog/butler/output/pubsub"
+        out "github.com/luci/luci-go/client/internal/logdog/butler/output/pubsub"
         "github.com/luci/luci-go/common/flag/multiflag"
         ps "github.com/luci/luci-go/common/gcloud/pubsub"
         log "github.com/luci/luci-go/common/logging"
+        "github.com/luci/luci-go/common/retry"
+        "golang.org/x/net/context"
+        "google.golang.org/cloud"
+        "google.golang.org/cloud/pubsub"
 )
 
 func init() {
         registerOutputFactory(new(pubsubOutputFactory))
 }
 
 // pubsubOutputFactory for Google Cloud PubSub.
 type pubsubOutputFactory struct {
         topic      ps.Topic
         noCompress bool
@@ skipping 22 matching lines @@
         if err := f.topic.Validate(); err != nil {
                 return nil, fmt.Errorf("pubsub: invalid topic name: %s", err)
         }
 
         // Instantiate our Pub/Sub instance. We will use the non-cancelling context,
         // as we want Pub/Sub system to drain without interruption if the application
         // is otherwise interrupted.
         ctx := log.SetFields(a.ncCtx, log.Fields{
                 "topic": f.topic,
         })
-        client, err := a.authenticatedClient(ctx)
+        ts, err := a.authenticatedTokenSource(ctx)
         if err != nil {
-                return nil, fmt.Errorf("pubsub: failed to initialize Pub/Sub context: %s", err)
+                return nil, fmt.Errorf("pubsub: failed to initialize Pub/Sub token source: %s", err)
         }
-        psConn := &ps.Retry{
-                Connection: ps.NewConnection(client),
-                Callback: func(err error, d time.Duration) {
-                        log.Fields{
-                                log.ErrorKey: err,
-                                "delay":      d,
-                        }.Warningf(ctx, "Transient error during Pub/Sub operation; retrying...")
-                },
+
+        // Split topic into Pub/Sub project and name.
+        project, name := f.topic.Split()
+
+        psClient, err := pubsub.NewClient(ctx, project, cloud.WithTokenSource(ts))
+        if err != nil {
+                return nil, fmt.Errorf("pubsub: failed to get Pub/Sub client: %s", err)
+        }
+        psTopic, err := psClient.NewTopic(ctx, name)
+        if err != nil {
+                return nil, fmt.Errorf("pubsub: failed to initialize Pub/Sub topic: %s", err)
         }
 
         // Assert that our Topic exists.
-        exists, err := psConn.TopicExists(ctx, f.topic)
+        exists, err := retryTopicExists(ctx, psTopic)
         if err != nil {
                 log.WithError(err).Errorf(ctx, "Failed to check for topic.")
                 return nil, err
         }
         if !exists {
                 log.Fields{
                         "topic": f.topic,
                 }.Errorf(ctx, "Pub/Sub Topic does not exist.")
                 return nil, fmt.Errorf("pubsub: topic %q does not exist", f.topic)
         }
 
-        return pubsub.New(ctx, pubsub.Config{
-                Publisher: psConn,
-                Topic:     f.topic,
-                Compress:  !f.noCompress,
-                Track:     f.track,
+        return out.New(ctx, out.Config{
+                Topic:    psTopic,
+                Compress: !f.noCompress,
+                Track:    f.track,
         }), nil
 }
+
+func retryTopicExists(ctx context.Context, t *pubsub.TopicHandle) (bool, error) {
+        var exists bool
+        err := retry.Retry(ctx, retry.Default, func() (err error) {
+                exists, err = t.Exists(ctx)
+                return
+        }, func(err error, d time.Duration) {
+                log.Fields{
+                        log.ErrorKey: err,
+                        "delay":      d,
+                }.Errorf(ctx, "Failed to check if topic exists; retrying...")
+        })
+        return exists, err
+}