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

Remove duplicated function for getShardIterator

Browse source
This commit is contained in:
Harlow Ward 2017-10-15 17:45:38 -07:00
parent c04f3d8a94
commit e4efcdb56e
1 changed files with 3 additions and 25 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

28
consumer.go
View file

@ -25,6 +25,7 @@ func NewConsumer(config Config) *Consumer {
} }
} }
// Consumer wraps the interaction with the Kinesis stream
type Consumer struct { type Consumer struct {
svc *kinesis.Kinesis svc *kinesis.Kinesis
Config Config
@ -102,33 +103,10 @@ func (c *Consumer) getShardIterator(shardID string) *string {
} }
if c.Checkpoint.CheckpointExists(shardID) { if c.Checkpoint.CheckpointExists(shardID) {
params.ShardIteratorType = aws.String(string(ShardIteratorAfterSequenceNumber)) params.ShardIteratorType = aws.String("AFTER_SEQUENCE_NUMBER")
params.StartingSequenceNumber = aws.String(c.Checkpoint.SequenceNumber()) params.StartingSequenceNumber = aws.String(c.Checkpoint.SequenceNumber())
} else { } else {
params.ShardIteratorType = aws.String(string(c.ShardIteratorType)) params.ShardIteratorType = aws.String("TRIM_HORIZON")
}
resp, err := c.svc.GetShardIterator(params)
if err != nil {
c.Logger.WithError(err).Error("GetShardIterator")
os.Exit(1)
}
return resp.ShardIterator
}
func (c *Consumer) getShardIterator(shardID string) *string {
params := &kinesis.GetShardIteratorInput{
ShardId: aws.String(shardID),
StreamName: aws.String(c.StreamName),
}
if c.Checkpoint.CheckpointExists(shardID) {
params.ShardIteratorType = aws.String(string(ShardIteratorAfterSequenceNumber))
params.StartingSequenceNumber = aws.String(c.Checkpoint.SequenceNumber())
} else {
params.ShardIteratorType = aws.String(string(c.ShardIteratorType))
} }
resp, err := c.svc.GetShardIterator(params) resp, err := c.svc.GetShardIterator(params)