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

Have new func return type

Browse source
This commit is contained in:
Harlow Ward 2017-11-26 18:22:09 -08:00
parent e5e057d6aa
commit 955f74d553
2 changed files with 3 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
client.go
View file

@ -9,8 +9,8 @@ import (
"github.com/aws/aws-sdk-go/service/kinesis" "github.com/aws/aws-sdk-go/service/kinesis"
) )
// NewKinesisClient returns a new wrapper around the Kinesis client // NewKinesisClient returns client to interface with Kinesis stream
func NewKinesisClient() Client { func NewKinesisClient() *KinesisClient {
svc := kinesis.New(session.New(aws.NewConfig())) svc := kinesis.New(session.New(aws.NewConfig()))
return &KinesisClient{svc} return &KinesisClient{svc}
} }

2
consumer.go
View file

@ -156,7 +156,7 @@ func (c *Consumer) Scan(ctx context.Context, fn func(*Record) bool) error {
// ScanShard loops over records on a specific shard, calls the callback func // ScanShard loops over records on a specific shard, calls the callback func
// for each record and checkpoints the progress of scan. // for each record and checkpoints the progress of scan.
// Note: Returning `false` from the callback func will end the scan. // Note: Returning `false` from the callback func will end the scan.
func (c *Consumer) ScanShard(ctx context.Context, shardID string, fn func(*Record) bool) (err error) { func (c *Consumer) ScanShard(ctx context.Context, shardID string, fn func(*Record) bool) (err error) {
lastSeqNum, err := c.checkpoint.Get(c.streamName, shardID) lastSeqNum, err := c.checkpoint.Get(c.streamName, shardID)
if err != nil { if err != nil {
return fmt.Errorf("get checkpoint error: %v", err) return fmt.Errorf("get checkpoint error: %v", err)