LogDog: Add collector service implementation.

client/internal/logdog/butler/output/pubsub/pubsubOutput_test.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"
         "compress/zlib"
         "errors"
         "fmt"
         "io/ioutil"
         "sync"
         "testing"
         "time"
 
         "github.com/golang/protobuf/proto"
         "github.com/luci/luci-go/common/clock"
         "github.com/luci/luci-go/common/clock/testclock"
         "github.com/luci/luci-go/common/gcloud/gcps"
         "github.com/luci/luci-go/common/logdog/protocol"
         "github.com/luci/luci-go/common/proto/google"
         "github.com/luci/luci-go/common/recordio"
         . "github.com/smartystreets/goconvey/convey"
         "golang.org/x/net/context"
         "google.golang.org/cloud/pubsub"
 )
 
-type testPubSub struct {
+type testPublisher struct {
         sync.Mutex
 
         err   error
         topic gcps.Topic
 
         msgC          chan *pubsub.Message
         nextMessageID int
 }
 
-func (*testPubSub) TopicExists(gcps.Topic) (bool, error)                   { panic("not implemented") }
-func (*testPubSub) SubExists(gcps.Subscription) (bool, error)              { panic("not implemented") }
-func (*testPubSub) Pull(gcps.Subscription, int) ([]*pubsub.Message, error) { panic("not implemented") }
-func (*testPubSub) Ack(gcps.Subscription, ...string) error                 { panic("not implemented") }
-
-func (ps *testPubSub) Publish(t gcps.Topic, msgs ...*pubsub.Message) ([]string, error) {
+func (ps *testPublisher) Publish(c context.Context, t gcps.Topic, msgs ...*pubsub.Message) ([]string, error) {
         if ps.err != nil {
                 return nil, ps.err
         }
         if t != ps.topic {
                 return nil, fmt.Errorf("test: published topic doesn't match configured (%s != %s)", t, ps.topic)
         }
 
         ids := make([]string, len(msgs))
         for i, m := range msgs {
                 ps.msgC <- m
                 ids[i] = ps.getNextMessageID()
         }
         return ids, nil
 }
 
-func (ps *testPubSub) getNextMessageID() string {
+func (ps *testPublisher) getNextMessageID() string {
         ps.Lock()
         defer ps.Unlock()
 
         id := ps.nextMessageID
         ps.nextMessageID++
         return fmt.Sprintf("%d", id)
 }
 
 func TestConfig(t *testing.T) {
         Convey(`A configuration instance`, t, func() {
-                ps := &testPubSub{}
+                ps := &testPublisher{}
                 conf := Config{
-                        PubSub: ps,
-                        Topic:  gcps.Topic("test-topic"),
+                        Publisher: ps,
+                        Topic:     gcps.Topic("test-topic"),
                 }
 
                 Convey(`Will successfully validate.`, func() {
                         So(conf.Validate(), ShouldBeNil)
                 })
 
                 Convey(`Will not validate without a PubSub instance.`, func() {
-                        conf.PubSub = nil
+                        conf.Publisher = nil
                         So(conf.Validate(), ShouldNotBeNil)
                 })
 
                 Convey(`Will not validate with an empty Topic.`, func() {
                         conf.Topic = ""
                         So(conf.Validate(), ShouldNotBeNil)
                 })
 
                 Convey(`Will not validate with an invalid Topic.`, func() {
                         conf.Topic = gcps.Topic("a!")
@@ skipping 45 matching lines @@
         if err := proto.Unmarshal(data, &dataBundle); err != nil {
                 return nil, nil, fmt.Errorf("test: failed to unmarshal bundle: %s", err)
         }
 
         return &header, &dataBundle, nil
 }
 
 func TestOutput(t *testing.T) {
         Convey(`An Output using a test Pub/Sub instance`, t, func() {
                 ctx, _ := testclock.UseTime(context.Background(), time.Date(2015, 1, 1, 0, 0, 0, 0, time.UTC))
-                ps := &testPubSub{
+                ps := &testPublisher{
                         topic: gcps.Topic("test-topic"),
                         msgC:  make(chan *pubsub.Message, 1),
                 }
                 conf := Config{
-                        PubSub: ps,
-                        Topic:  gcps.Topic("test-topic"),
+                        Publisher: ps,
+                        Topic:     gcps.Topic("test-topic"),
                 }
                 o := New(ctx, conf).(*gcpsOutput)
                 So(o, ShouldNotBeNil)
                 defer o.Close()
 
                 bundle := &protocol.ButlerLogBundle{
                         Source:    "GCPS Test",
                         Timestamp: google.NewTimestamp(clock.Now(ctx)),
                         Entries: []*protocol.ButlerLogBundle_Entry{
                                 {},
@@ skipping 16 matching lines @@
                                 So(st.DiscardedMessages(), ShouldEqual, 0)
                         })
                 })
 
                 Convey(`Will return an error if Publish failed.`, func() {
                         ps.err = errors.New("test: error")
                         So(o.SendBundle(bundle), ShouldNotBeNil)
                 })
         })
 }