LogDog: Add collector service implementation.

common/gcloud/gcps/pubsub_impl.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 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"
 )
 
 // pubSubImpl is an implementation of PubSub that communicates directly to the
 // Google Cloud Pub/Sub system.
 //
 // 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)
+}