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

Run initial scan immediately (#123)

Browse source 
Rather than starting the shard scan loop and waiting for next scan tick before
fetching any data, do the first poll immediately, and then wait.
This commit is contained in:
Jason Tackaberry 2020-08-01 18:45:37 -04:00 committed by GitHub
parent 97ffabeaa5
commit 3f2519e51c
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
1 changed files with 23 additions and 22 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

45
consumer.go
View file

@ -158,33 +158,26 @@ func (c *Consumer) ScanShard(ctx context.Context, shardID string, fn ScanFunc) e
defer scanTicker.Stop() defer scanTicker.Stop()
for { for {
select { resp, err := c.client.GetRecords(&kinesis.GetRecordsInput{
case <-ctx.Done(): Limit: aws.Int64(c.maxRecords),
return nil ShardIterator: shardIterator,
case <-scanTicker.C: })
resp, err := c.client.GetRecords(&kinesis.GetRecordsInput{
Limit: aws.Int64(c.maxRecords),
ShardIterator: shardIterator,
})
// attempt to recover from GetRecords error when expired iterator // attempt to recover from GetRecords error when expired iterator
if err != nil { if err != nil {
c.logger.Log("[CONSUMER] get records error:", err.Error()) c.logger.Log("[CONSUMER] get records error:", err.Error())
if awserr, ok := err.(awserr.Error); ok { if awserr, ok := err.(awserr.Error); ok {
if _, ok := retriableErrors[awserr.Code()]; !ok { if _, ok := retriableErrors[awserr.Code()]; !ok {
return fmt.Errorf("get records error: %v", awserr.Message()) return fmt.Errorf("get records error: %v", awserr.Message())
}
} }
shardIterator, err = c.getShardIterator(ctx, c.streamName, shardID, lastSeqNum)
if err != nil {
return fmt.Errorf("get shard iterator error: %v", err)
}
continue
} }
shardIterator, err = c.getShardIterator(ctx, c.streamName, shardID, lastSeqNum)
if err != nil {
return fmt.Errorf("get shard iterator error: %v", err)
}
} else {
// loop over records, call callback func // loop over records, call callback func
for _, r := range resp.Records { for _, r := range resp.Records {
select { select {
@ -214,6 +207,14 @@ func (c *Consumer) ScanShard(ctx context.Context, shardID string, fn ScanFunc) e
shardIterator = resp.NextShardIterator shardIterator = resp.NextShardIterator
} }
// Wait for next scan
select {
case <-ctx.Done():
return nil
case <-scanTicker.C:
continue
}
} }
} }