LogDog: Add collector service implementation.

common/gcloud/gcps/retry.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 (
+        "github.com/luci/luci-go/common/retry"
+        "golang.org/x/net/context"
+        "google.golang.org/cloud/pubsub"
+)
+
+// Retry wraps a PubSub implementation and retries on transient errors.

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

This is super useful. Now we don't have to have one-off retry wrappers every
time someone wants to use Pub/Sub...

+type Retry struct {
+        // PS is the PubSub to retry.
+        PS PubSub
+        // G is the retry generator to use. If nil, retry.Default will be used.
+        G retry.Generator
+
+        // C, if not nil, will be called when an error is encountered.
+        C retry.Callback
+}
+
+// TopicExists implements the PubSub interface.
+func (r *Retry) TopicExists(c context.Context, t Topic) (exists bool, err error) {
+        err = retry.Retry(c, r.retryIter(), func() (err error) {
+                exists, err = r.TopicExists(c, t)
+                return
+        }, r.C)
+        return
+}
+
+// SubExists implements the PubSub interface.
+func (r *Retry) SubExists(c context.Context, s Subscription) (exists bool, err error) {
+        err = retry.Retry(c, r.retryIter(), func() (err error) {
+                exists, err = r.SubExists(c, s)
+                return
+        }, r.C)
+        return
+}
+
+// Publish implements the PubSub interface.
+func (r *Retry) Publish(c context.Context, t Topic, msgs ...*pubsub.Message) (ids []string, err error) {
+        err = retry.Retry(c, r.retryIter(), func() (err error) {
+                ids, err = r.Publish(c, t, msgs...)
+                return
+        }, r.C)
+        return
+}
+
+// Pull implements the PubSub interface.
+func (r *Retry) Pull(c context.Context, s Subscription, batch int) (msgs []*pubsub.Message, err error) {
+        err = retry.Retry(c, r.retryIter(), func() (err error) {
+                msgs, err = r.Pull(c, s, batch)
+                return
+        }, r.C)
+        return
+}
+
+// Ack implements the PubSub interface.
+func (r *Retry) Ack(c context.Context, s Subscription, ackIDs ...string) (err error) {
+        return retry.Retry(c, r.retryIter(), func() error {
+                return r.Ack(c, s, ackIDs...)
+        }, r.C)
+}
+
+func (r *Retry) retryIter() retry.Iterator {
+        var it retry.Iterator
+        if r.G != nil {
+                it = r.G()
+        } else {
+                it = retry.Default()
+        }
+        return retry.TransientOnly(it)
+}