LogDog: Add collector service implementation.

client/internal/logdog/butler/output/pubsub/pubsubOutput.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 pubsub
 
 import (
         "bytes"
         "errors"
         "fmt"
         "sync"
         "time"
 
         "github.com/luci/luci-go/client/internal/logdog/butler/output"
         "github.com/luci/luci-go/common/gcloud/gcps"
         "github.com/luci/luci-go/common/logdog/butlerproto"
         "github.com/luci/luci-go/common/logdog/protocol"
         log "github.com/luci/luci-go/common/logging"
         "github.com/luci/luci-go/common/recordio"
         "github.com/luci/luci-go/common/retry"
         "golang.org/x/net/context"
         "google.golang.org/cloud/pubsub"
 )
 
+// Publisher is an interface for something that publishes Pub/Sub messages.

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

Make a specific interface instead of requiring one that implements all methods
when we only need Publish :)

+//
+// gcps.PubSub implements this interface.
+type Publisher interface {
+        // Publish mirrors the gcps.PubSub Publish method.
+        Publish(context.Context, gcps.Topic, ...*pubsub.Message) ([]string, error)
+}
+
+var _ Publisher = gcps.PubSub(nil)
+
 // Config is a configuration structure for GCPS output.
 type Config struct {
-        // Pubsub is the Pub/Sub instance to use.
-        PubSub gcps.PubSub
+        // Publisher is the Pub/Sub instance to use.
+        Publisher Publisher
 
         // Topic is the name of the Cloud Pub/Sub topic to publish to.
         Topic gcps.Topic
 
         // Compress, if true, enables zlib compression.
         Compress bool
 }
 
 // Validate validates the Output configuration.
 func (c *Config) Validate() error {
-        if c.PubSub == nil {
+        if c.Publisher == nil {
                 return errors.New("gcps: no pub/sub instance configured")
         }
         if err := c.Topic.Validate(); err != nil {
                 return fmt.Errorf("gcps: invalid Topic [%s]: %s", c.Topic, err)
         }
         return nil
 }
 
 // gcpsBuffer
 type gcpsBuffer struct {
@@ skipping 108 matching lines @@
                 Data: buf.Bytes(),
         }, nil
 }
 
 // publishMessages handles an individual publish request. It will indefinitely
 // retry transient errors until the publish succeeds.
 func (o *gcpsOutput) publishMessages(messages []*pubsub.Message) error {
         var messageIDs []string
         count := 0
         err := retry.Retry(o, retry.TransientOnly(retry.Default()), func() error {
-                ids, err := o.PubSub.Publish(o.Topic, messages...)
+                ids, err := o.Publisher.Publish(o, o.Topic, messages...)
                 if err != nil {
                         return err
                 }
                 messageIDs = ids
                 return nil
         }, func(err error, d time.Duration) {
                 log.Fields{
                         log.ErrorKey: err,
                         "count":      count,
                         "delay":      d,
                 }.Warningf(o, "Transient publish error; retrying.")
                 count++
         })
         if err != nil {
                 log.Errorf(log.SetError(o, err), "Failed to send PubSub message.")
                 return err
         }
 
         log.Debugf(log.SetField(o, "messageIds", messageIDs), "Published messages.")
         return nil
 }
 
 func (o *gcpsOutput) mergeStats(s output.Stats) {
         o.statsMu.Lock()
         defer o.statsMu.Unlock()
 
         o.stats.Merge(s)
 }