reduce: separate batch calls

1 files changed, 7 insertions, 2 deletions

diff --git a/skate/zipkey/batch.go b/skate/zipkey/batch.go
index ebbd081..85d1d35 100644
--- a/skate/zipkey/batch.go
+++ b/skate/zipkey/batch.go
@@ -19,9 +19,14 @@ type Batcher struct {
 
 // NewBatcher set ups a new Batcher.
 func NewBatcher(gf groupFunc) *Batcher {
+	return NewBatcherSize(gf, 1000)
+}
+
+// NewBatcherSize initializes a batcher with a given size.
+func NewBatcherSize(gf groupFunc, size int) *Batcher {
 	batcher := Batcher{
 		gf:         gf,
-		Size:       1000,
+		Size:       size,
 		NumWorkers: runtime.NumCPU(),
 		queue:      make(chan []*Group),
 	}
@@ -46,7 +51,7 @@ func (b *Batcher) Close() error {
 }
 
 // GroupFunc implement the groupFunc type. Not thread safe. Panics if called
-// after Close has been called.
+// after Close.
 func (b *Batcher) GroupFunc(g *Group) error {
 	if b.closing {
 		panic("cannot call GroupFunc after Close")