Add logging when records are emitted w/ record count

README.md

@@ -24,7 +24,7 @@ func main() {
   c := connector.NewConsumer(*app, *stream)
 
   // override default values
-  c.Set("maxBatchCount", 200)
+  c.Set("maxRecordCount", 200)
 
   // start consuming records from the queues
   c.Start(connector.HandlerFunc(func(b connector.Buffer) {

buffer.go

@@ -9,12 +9,12 @@ type Buffer struct {
 	firstSequenceNumber string
 	lastSequenceNumber  string
 
-	MaxBatchCount int
+	MaxRecordCount int
 }
 
 // AddRecord adds a record to the buffer.
 func (b *Buffer) AddRecord(r *kinesis.Record) {
-	if len(b.records) == 0 {
+	if b.RecordCount() == 0 {
 		b.firstSequenceNumber = *r.SequenceNumber
 	}
 
@@ -24,7 +24,7 @@ func (b *Buffer) AddRecord(r *kinesis.Record) {
 
 // ShouldFlush determines if the buffer has reached its target size.
 func (b *Buffer) ShouldFlush() bool {
-	return len(b.records) >= b.MaxBatchCount
+	return b.RecordCount() >= b.MaxRecordCount
 }
 
 // Flush empties the buffer and resets the sequence counter.
@@ -37,6 +37,11 @@ func (b *Buffer) GetRecords() []*kinesis.Record {
 	return b.records
 }
 
+// RecordCount returns the number of records in the buffer.
+func (b *Buffer) RecordCount() int {
+	return len(b.records)
+}
+
 // FirstSequenceNumber returns the sequence number of the first record in the buffer.
 func (b *Buffer) FirstSeq() string {
 	return b.firstSequenceNumber

consumer.go

@@ -11,7 +11,8 @@ import (
 )
 
 var (
-	maxBatchCount = 1000
+	maxRecordCount = 1000
+	maxBufferTime  = "30s"
 )
 
 // NewConsumer creates a new kinesis connection and returns a
@@ -41,8 +42,8 @@ type Consumer struct {
 // Set `option` to `value`
 func (c *Consumer) Set(option string, value interface{}) {
 	switch option {
-	case "maxBatchCount":
-		maxBatchCount = value.(int)
+	case "maxRecordCount":
+		maxRecordCount = value.(int)
 	default:
 		log.Error("invalid option")
 		os.Exit(1)
@@ -81,7 +82,7 @@ func (c *Consumer) handlerLoop(shardID string, handler Handler) {
 	})
 
 	buf := &Buffer{
-		MaxBatchCount: maxBatchCount,
+		MaxRecordCount: maxRecordCount,
 	}
 
 	checkpoint := &Checkpoint{
@@ -108,7 +109,7 @@ func (c *Consumer) handlerLoop(shardID string, handler Handler) {
 	}
 
 	shardIterator := resp.ShardIterator
-	ctx.Info("started")
+	ctx.Info("processing")
 
 	for {
 		resp, err := c.svc.GetRecords(
@@ -128,6 +129,7 @@ func (c *Consumer) handlerLoop(shardID string, handler Handler) {
 
 				if buf.ShouldFlush() {
 					handler.HandleRecords(*buf)
+					ctx.WithField("count", buf.RecordCount()).Info("emitted")
 					checkpoint.SetCheckpoint(shardID, buf.LastSeq())
 					buf.Flush()
 				}

consumer_test.go

@@ -7,11 +7,11 @@ import (
 )
 
 func Test_Set(t *testing.T) {
-	defaultMaxBatchCount := 1000
-	assert.Equal(t, maxBatchCount, defaultMaxBatchCount)
+	defaultMaxRecordCount := 1000
+	assert.Equal(t, maxRecordCount, defaultMaxRecordCount)
 
 	c := NewConsumer("app", "stream")
-	c.Set("maxBatchCount", 100)
+	c.Set("maxRecordCount", 100)
 
-	assert.Equal(t, maxBatchCount, 100)
+	assert.Equal(t, maxRecordCount, 100)
 }

examples/firehose/main.go

@@ -32,8 +32,7 @@ func main() {
 	svc := firehose.New(session.New())
 
 	c := connector.NewConsumer(*app, *stream)
-	c.Set("maxBatchCount", 400)
-	c.Set("pollInterval", "3s")
+	c.Set("maxRecordCount", 400)
 	c.Start(connector.HandlerFunc(func(b connector.Buffer) {
 		params := &firehose.PutRecordBatchInput{
 			DeliveryStreamName: aws.String(*delivery),