LogDog: Add collector service implementation.

common/gcloud/gcps/pubsub_impl.go

Index: common/gcloud/gcps/pubsub_impl.go
diff --git a/common/gcloud/gcps/pubsub_impl.go b/common/gcloud/gcps/pubsub_impl.go
index cc9c3b3ea9fa5ba70c2a3766013163ee5d64a3b9..0209018c59758180c3bf85f28434b714e2c53cc9 100644
--- a/common/gcloud/gcps/pubsub_impl.go
+++ b/common/gcloud/gcps/pubsub_impl.go
@@ -5,8 +5,11 @@
 package gcps
 
 import (
+	"net/http"
+
 	"github.com/luci/luci-go/common/errors"
 	"golang.org/x/net/context"
+	"google.golang.org/cloud"
 	"google.golang.org/cloud/pubsub"
 )
 
@@ -15,40 +18,46 @@ import (
 //
 // Currently, all errors are regarded as transient.
 type pubSubImpl struct {
-	ctx context.Context
+	client  *http.Client

[dnj (Google), 2016/01/21 04:36:24]

Rearrange to accommodate the end-method Context change.

+	project string
 }
 
 // New instantiates a new PubSub instance configured to use the Google Cloud
 // Pub/Sub system.
 //
-// The supplied context must be properly authenticated to interface with the
+// The supplied Client must be properly authenticated to interface with the
 // named Pub/Sub system.
-func New(ctx context.Context) PubSub {
+func New(c *http.Client, project string) PubSub {
 	return &pubSubImpl{
-		ctx: ctx,
+		client:  c,
+		project: project,
 	}
 }
 
-func (p *pubSubImpl) TopicExists(t Topic) (bool, error) {
-	exists, err := pubsub.TopicExists(p.ctx, string(t))
+func (p *pubSubImpl) TopicExists(c context.Context, t Topic) (bool, error) {
+	exists, err := pubsub.TopicExists(p.with(c), string(t))
 	return exists, (err)
 }
 
-func (p *pubSubImpl) SubExists(s Subscription) (bool, error) {
-	exists, err := pubsub.SubExists(p.ctx, string(s))
+func (p *pubSubImpl) SubExists(c context.Context, s Subscription) (bool, error) {
+	exists, err := pubsub.SubExists(p.with(c), string(s))
 	return exists, errors.WrapTransient(err)
 }
 
-func (p *pubSubImpl) Publish(t Topic, msgs ...*pubsub.Message) ([]string, error) {
-	ids, err := pubsub.Publish(p.ctx, string(t), msgs...)
+func (p *pubSubImpl) Publish(c context.Context, t Topic, msgs ...*pubsub.Message) ([]string, error) {
+	ids, err := pubsub.Publish(p.with(c), string(t), msgs...)
 	return ids, errors.WrapTransient(err)
 }
 
-func (p *pubSubImpl) Pull(s Subscription, n int) ([]*pubsub.Message, error) {
-	msgs, err := pubsub.Pull(p.ctx, string(s), n)
+func (p *pubSubImpl) Pull(c context.Context, s Subscription, n int) ([]*pubsub.Message, error) {
+	msgs, err := pubsub.Pull(p.with(c), string(s), n)
 	return msgs, errors.WrapTransient(err)
 }
 
-func (p *pubSubImpl) Ack(s Subscription, ackIDs ...string) error {
-	return errors.WrapTransient(pubsub.Ack(p.ctx, string(s), ackIDs...))
+func (p *pubSubImpl) Ack(c context.Context, s Subscription, ackIDs ...string) error {
+	return errors.WrapTransient(pubsub.Ack(p.with(c), string(s), ackIDs...))
+}
+
+func (p *pubSubImpl) with(c context.Context) context.Context {
+	return cloud.WithContext(c, p.project, p.client)
 }