LogDog: Add collector service implementation.

common/gcloud/gcps/ackbuffer/ackbuffer_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 ackbuffer
 
 import (
-        "fmt"
         "sort"
         "sync"
         "testing"
         "time"
 
         "github.com/luci/luci-go/common/clock"
         "github.com/luci/luci-go/common/clock/testclock"
         "github.com/luci/luci-go/common/errors"
-        "github.com/luci/luci-go/common/gcloud/gcps"
         "github.com/luci/luci-go/common/stringset"
         "golang.org/x/net/context"
 
         . "github.com/smartystreets/goconvey/convey"
 )
 
-type testPubSub struct {
+type testACK struct {
         sync.Mutex
 
-        err  error
-        sub  gcps.Subscription
-        acks stringset.Set
+        err       error
+        acks      stringset.Set
+        batchSize int
 }
 
-func (ps *testPubSub) Ack(s gcps.Subscription, acks ...string) error {
+func (ps *testACK) Ack(c context.Context, acks ...string) error {
         ps.Lock()
         defer ps.Unlock()
 
         if ps.err != nil {
                 return ps.err
         }
 
-        if s != ps.sub {
-                return fmt.Errorf("unknown subscription %q", s)
-        }
-
         if ps.acks == nil {
                 ps.acks = stringset.New(0)
         }
         for _, ack := range acks {
                 ps.acks.Add(ack)
         }
         return nil
 }
 
-func (ps *testPubSub) ackIDs() []string {
+func (ps *testACK) AckBatchSize() int {
+        size := ps.batchSize
+        if size <= 0 {
+                size = 4
+        }
+        return size
+}
+
+func (ps *testACK) ackIDs() []string {
         ps.Lock()
         defer ps.Unlock()
 
         v := make([]string, 0, ps.acks.Len())
         ps.acks.Iter(func(s string) bool {
                 v = append(v, s)
                 return true
         })
         sort.Strings(v)
         return v
 }
 
 func TestAckBuffer(t *testing.T) {
         t.Parallel()
 
         Convey(`An AckBuffer configuration using a testing Pub/Sub`, t, func() {
                 c := context.Background()
                 c, tc := testclock.UseTime(c, testclock.TestTimeLocal)
-                ps := &testPubSub{
-                        sub: gcps.Subscription("testsub"),
-                }
+                ps := &testACK{}
 
                 var discarded []string
                 var discardedMu sync.Mutex
 
                 cfg := Config{
-                        PubSub:       ps,
-                        Subscription: ps.sub,
+                        Ack: ps,
                         DiscardCallback: func(acks []string) {
                                 discardedMu.Lock()
                                 defer discardedMu.Unlock()
 
                                 discarded = append(discarded, acks...)
                         },
                 }
 
                 Convey(`Can instantiate an AckBuffer`, func() {
                         ab := New(c, cfg)
@@ skipping 39 matching lines @@
                                         closeOnce()
 
                                         sort.Strings(acks)
                                         sort.Strings(discarded)
                                         So(discarded, ShouldResemble, acks)
                                 })
                         })
                 })
         })
 }