luciano/kinesis-consumer
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Increment counts in correct usage points

Browse source
This commit is contained in:
Harlow Ward 2017-11-22 14:21:19 -08:00
parent c91f6233ef
commit c64b40b4ad
1 changed files with 2 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
consumer.go
View file

@ -185,18 +185,17 @@ loop:
break loop break loop
default: default:
lastSeqNum = *r.SequenceNumber lastSeqNum = *r.SequenceNumber
c.counter.Add("records", 1)
if ok := fn(r); !ok { if ok := fn(r); !ok {
break loop break loop
} }
} }
} }
c.counter.Add("records", int64(len(resp.Records)))
c.counter.Add("checkpoints", 1)
if err := c.checkpoint.Set(shardID, lastSeqNum); err != nil { if err := c.checkpoint.Set(shardID, lastSeqNum); err != nil {
c.logger.Printf("set checkpoint error: %v", err) c.logger.Printf("set checkpoint error: %v", err)
} }
c.counter.Add("checkpoints", 1)
} }
if resp.NextShardIterator == nil || shardIterator == resp.NextShardIterator { if resp.NextShardIterator == nil || shardIterator == resp.NextShardIterator {