Change Batcher.Start() to return channels

This is more useful for client code, and is a more common Go pattern. As
a result the goroutine is launched within start.

batcher.go

@@ -27,38 +27,46 @@ type PointBatcherStats struct {
 	TimeoutTotal uint64 // Nubmer of timeouts that occurred.
 }
 
-// Start starts the batching process. It should be called from a goroutine.
-func (b *PointBatcher) Start(in <-chan Point, out chan<- []Point) {
+// Start starts the batching process. Returns the in and out channels for points
+// batch point-batches respectively.
+func (b *PointBatcher) Start() (chan<- Point, <-chan []Point) {
 	var timer *time.Timer
 	var batch []Point
 	var timerCh <-chan time.Time
 
-	for {
-		select {
-		case p := <-in:
-			atomic.AddUint64(&b.stats.PointTotal, 1)
-			if batch == nil {
-				batch = make([]Point, 0, b.size)
-				timer = time.NewTimer(b.duration)
-				timerCh = timer.C
-			}
+	in := make(chan Point)
+	out := make(chan []Point)
+
+	go func() {
+		for {
+			select {
+			case p := <-in:
+				atomic.AddUint64(&b.stats.PointTotal, 1)
+				if batch == nil {
+					batch = make([]Point, 0, b.size)
+					timer = time.NewTimer(b.duration)
+					timerCh = timer.C
+				}
+
+				batch = append(batch, p)
+				if len(batch) >= b.size { // 0 means send immediately.
+					atomic.AddUint64(&b.stats.SizeTotal, 1)
+					out <- batch
+					atomic.AddUint64(&b.stats.BatchTotal, 1)
+					batch = nil
+					timerCh = nil
+				}
 
-			batch = append(batch, p)
-			if len(batch) >= b.size { // 0 means send immediately.
-				atomic.AddUint64(&b.stats.SizeTotal, 1)
+			case <-timerCh:
+				atomic.AddUint64(&b.stats.TimeoutTotal, 1)
 				out <- batch
 				atomic.AddUint64(&b.stats.BatchTotal, 1)
 				batch = nil
-				timerCh = nil
 			}
-
-		case <-timerCh:
-			atomic.AddUint64(&b.stats.TimeoutTotal, 1)
-			out <- batch
-			atomic.AddUint64(&b.stats.BatchTotal, 1)
-			batch = nil
 		}
-	}
+	}()
+
+	return in, out
 }
 
 // Stats returns a PointBatcherStats object for the PointBatcher. While the each statistic should be

batcher_test.go

@@ -13,10 +13,7 @@ func TestBatch_Size(t *testing.T) {
 		t.Fatal("failed to create batcher for size test")
 	}
 
-	in := make(chan Point)
-	out := make(chan []Point)
-
-	go batcher.Start(in, out)
+	in, out := batcher.Start()
 
 	var p Point
 	go func() {
@@ -39,10 +36,7 @@ func TestBatch_Timeout(t *testing.T) {
 		t.Fatal("failed to create batcher for timeout test")
 	}
 
-	in := make(chan Point)
-	out := make(chan []Point)
-
-	go batcher.Start(in, out)
+	in, out := batcher.Start()
 
 	var p Point
 	go func() {
@@ -65,10 +59,7 @@ func TestBatch_MultipleBatches(t *testing.T) {
 		t.Fatal("failed to create batcher for size test")
 	}
 
-	in := make(chan Point)
-	out := make(chan []Point)
-
-	go batcher.Start(in, out)
+	in, out := batcher.Start()
 
 	var p Point
 	var b []Point