LogDog: Add collector service implementation.

common/gcloud/gcps/subscriber/source.go

Index: common/gcloud/gcps/subscriber/source.go
diff --git a/common/gcloud/gcps/subscriber/source.go b/common/gcloud/gcps/subscriber/source.go
new file mode 100644
index 0000000000000000000000000000000000000000..95dc93a26f857b852085a57f7abafbbf75f13080
--- /dev/null
+++ b/common/gcloud/gcps/subscriber/source.go
@@ -0,0 +1,44 @@
+// Copyright 2016 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 subscriber
+
+import (
+	"github.com/luci/luci-go/common/gcloud/gcps"
+	"golang.org/x/net/context"
+	"google.golang.org/cloud/pubsub"
+)
+
+// Source is used to pull Pub/Sub messages in batches.
+type Source interface {

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

Generalized Source allows for greater interoperability. NewSource below wraps a
Pub/Sub instance, making it easy to get a Source instance.

+	// Pull retrieves a new batch of Pub/Sub messages to process.
+	Pull(context.Context) ([]*pubsub.Message, error)
+}
+
+// PubSubSource is a Source implementation built on top of a gcps.PubSub.
+type pubSubSource struct {
+	ps    gcps.PubSub
+	sub   gcps.Subscription
+	batch int
+}
+
+// NewSource generates a new Source by wrapping a gcps.PubSub method. This
+// Source is bound to a single subscription.
+//
+// If the supplied batch size is <= 0, the maximum allowed Pub/Sub batch size
+// will be used.
+func NewSource(ps gcps.PubSub, s gcps.Subscription, batch int) Source {
+	if batch <= 0 {
+		batch = gcps.MaxSubscriptionPullSize
+	}
+	return &pubSubSource{
+		ps:    ps,
+		sub:   s,
+		batch: batch,
+	}
+}
+
+func (s *pubSubSource) Pull(c context.Context) (msgs []*pubsub.Message, err error) {
+	return s.ps.Pull(c, s.sub, s.batch)
+}