3372ea1d17
When immediate rescan is enabled, a poll from Kinesis that returns records and also indicates we are still behind the latest record will skip waiting for the next scan interval and rescan immediately. Related, the initial scan for a shard is performed immediately. The scan ticker only applies to subsequent scans. These changes make higher scan intervals feasible, allowing for less chatty clients while also improving overall effective throughput.
83 lines
1.8 KiB
Go
83 lines
1.8 KiB
Go
package consumer
|
|
|
|
import (
|
|
"time"
|
|
|
|
"github.com/aws/aws-sdk-go/service/kinesis/kinesisiface"
|
|
)
|
|
|
|
// Option is used to override defaults when creating a new Consumer
|
|
type Option func(*Consumer)
|
|
|
|
// WithGroup overrides the default storage
|
|
func WithGroup(group Group) Option {
|
|
return func(c *Consumer) {
|
|
c.group = group
|
|
}
|
|
}
|
|
|
|
// WithStore overrides the default storage
|
|
func WithStore(store Store) Option {
|
|
return func(c *Consumer) {
|
|
c.store = store
|
|
}
|
|
}
|
|
|
|
// WithLogger overrides the default logger
|
|
func WithLogger(logger Logger) Option {
|
|
return func(c *Consumer) {
|
|
c.logger = logger
|
|
}
|
|
}
|
|
|
|
// WithCounter overrides the default counter
|
|
func WithCounter(counter Counter) Option {
|
|
return func(c *Consumer) {
|
|
c.counter = counter
|
|
}
|
|
}
|
|
|
|
// WithClient overrides the default client
|
|
func WithClient(client kinesisiface.KinesisAPI) Option {
|
|
return func(c *Consumer) {
|
|
c.client = client
|
|
}
|
|
}
|
|
|
|
// WithShardIteratorType overrides the starting point for the consumer
|
|
func WithShardIteratorType(t string) Option {
|
|
return func(c *Consumer) {
|
|
c.initialShardIteratorType = t
|
|
}
|
|
}
|
|
|
|
// WithTimestamp overrides the starting point for the consumer
|
|
func WithTimestamp(t time.Time) Option {
|
|
return func(c *Consumer) {
|
|
c.initialTimestamp = &t
|
|
}
|
|
}
|
|
|
|
// WithScanInterval overrides the scan interval for the consumer
|
|
func WithScanInterval(d time.Duration) Option {
|
|
return func(c *Consumer) {
|
|
c.scanInterval = d
|
|
}
|
|
}
|
|
|
|
// WithImmediateRescan overrides whether we wait for the next
|
|
// scan interval if records were fetched during a poll
|
|
func WithImmediateRescan(r bool) Option {
|
|
return func(c *Consumer) {
|
|
c.immediateRescan = r
|
|
}
|
|
}
|
|
|
|
// WithMaxRecords overrides the maximum number of records to be
|
|
// returned in a single GetRecords call for the consumer (specify a
|
|
// value of up to 10,000)
|
|
func WithMaxRecords(n int64) Option {
|
|
return func(c *Consumer) {
|
|
c.maxRecords = n
|
|
}
|
|
}
|