LogDog: BigTable batching schema.

common/gcloud/pubsub/subscriber/subscriber_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 subscriber
 
 import (
         "fmt"
         "sort"
         "sync"
@@ skipping 13 matching lines @@
 type event struct {
         msg *pubsub.Message
         err error
 }
 
 type testSource struct {
         sub    pubsub.Subscription
         eventC chan event
 }
 
-func (s *testSource) Pull(c context.Context) ([]*pubsub.Message, error) {
+func (s *testSource) Pull(c context.Context, batch int) ([]*pubsub.Message, error) {
         select {
         case <-c.Done():
                 return nil, c.Err()
 
         case e := <-s.eventC:
                 switch {
                 case e.err != nil:
                         return nil, e.err
 
                 case e.msg != nil:
@@ skipping 62 matching lines @@
                 c, tc := testclock.UseTime(c, testclock.TestTimeLocal)
 
                 c, cancelFunc := context.WithCancel(c)
                 defer cancelFunc()
 
                 src := &testSource{
                         eventC: make(chan event),
                 }
                 ack := &testACK{}
                 s := Subscriber{
-                        S:           src,
-                        A:           ack,
-                        PullWorkers: 8,
+                        S:       src,
+                        A:       ack,
+                        Workers: 8,
                 }
 
                 var seenMu sync.Mutex
                 seen := map[string]struct{}{}
                 blacklist := map[string]struct{}{}
                 runWith := func(cb func()) {
                         doneC := make(chan struct{})
                         go func() {
                                 defer close(doneC)
                                 s.Run(c, func(msg *pubsub.Message) bool {
@@ skipping 55 matching lines @@
                                 src.message("a", "b")
                                 src.error(errors.New("test error"))
                                 src.message("c", "d")
                         })
 
                         So(dumpStringSet(seen), ShouldResemble, []string{"a", "b", "c", "d"})
                         So(ack.getACKs(), ShouldResemble, []string{"a", "b", "c", "d"})
                 })
         })
 }