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

Include MillisBehindLatest in Record for ScanFunc (#124)

Browse source
This commit is contained in:
Jason Tackaberry 2020-08-02 01:05:17 -04:00 committed by GitHub
parent 3f2519e51c
commit e60d217333
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
1 changed files with 3 additions and 2 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
consumer.go
View file

@ -20,7 +20,8 @@ import (
// extends to include the shard id. // extends to include the shard id.
type Record struct { type Record struct {
*kinesis.Record *kinesis.Record
ShardID string ShardID string
MillisBehindLatest *int64
} }
// New creates a kinesis consumer with default settings. Use Option to override // New creates a kinesis consumer with default settings. Use Option to override
@ -184,7 +185,7 @@ func (c *Consumer) ScanShard(ctx context.Context, shardID string, fn ScanFunc) e
case <-ctx.Done(): case <-ctx.Done():
return nil return nil
default: default:
err := fn(&Record{r, shardID}) err := fn(&Record{r, shardID, resp.MillisBehindLatest})
if err != nil && err != ErrSkipCheckpoint { if err != nil && err != ErrSkipCheckpoint {
return err return err
} }