LogDog: Add collector service implementation.

common/gcloud/gcps/retry.go

Index: common/gcloud/gcps/retry.go
diff --git a/common/gcloud/gcps/retry.go b/common/gcloud/gcps/retry.go
new file mode 100644
index 0000000000000000000000000000000000000000..53830bd00f11580c29aebc578a9ca361ef657aac
--- /dev/null
+++ b/common/gcloud/gcps/retry.go
@@ -0,0 +1,75 @@
+// 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)
+}