kinesumer

Jul 14, 2022

Changes in this version

+ var ErrEmptySequenceNumber = errors.New("kinesumer: sequence number can't be empty")

+ var ErrInvalidStream = errors.New("kinesumer: invalid stream")

+ type CommitConfig struct

+ Auto bool

+ Interval time.Duration

+ Timeout time.Duration

+ func NewDefaultCommitConfig() *CommitConfig

type Config

+ Commit *CommitConfig

type Kinesumer

+ func (k *Kinesumer) Commit()

+ func (k *Kinesumer) MarkRecord(record *Record)

+ type MockStateStore struct

+ func NewMockStateStore(ctrl *gomock.Controller) *MockStateStore

+ func (m *MockStateStore) DeregisterClient(ctx context.Context, clientID string) error

+ func (m *MockStateStore) EXPECT() *MockStateStoreMockRecorder

+ func (m *MockStateStore) GetShards(ctx context.Context, stream string) (Shards, error)

+ func (m *MockStateStore) ListAllAliveClientIDs(ctx context.Context) ([]string, error)

+ func (m *MockStateStore) ListCheckPoints(ctx context.Context, stream string, shardIDs []string) (map[string]string, error)

+ func (m *MockStateStore) PingClientAliveness(ctx context.Context, clientID string) error

+ func (m *MockStateStore) PruneClients(ctx context.Context) error

+ func (m *MockStateStore) RegisterClient(ctx context.Context, clientID string) error

+ func (m *MockStateStore) UpdateCheckPoints(ctx context.Context, checkpoints []*ShardCheckPoint) error

+ func (m *MockStateStore) UpdateShards(ctx context.Context, stream string, shards Shards) error

+ type MockStateStoreMockRecorder struct

+ func (mr *MockStateStoreMockRecorder) DeregisterClient(ctx, clientID interface{}) *gomock.Call

+ func (mr *MockStateStoreMockRecorder) GetShards(ctx, stream interface{}) *gomock.Call

+ func (mr *MockStateStoreMockRecorder) ListAllAliveClientIDs(ctx interface{}) *gomock.Call

+ func (mr *MockStateStoreMockRecorder) ListCheckPoints(ctx, stream, shardIDs interface{}) *gomock.Call

+ func (mr *MockStateStoreMockRecorder) PingClientAliveness(ctx, clientID interface{}) *gomock.Call

+ func (mr *MockStateStoreMockRecorder) PruneClients(ctx interface{}) *gomock.Call

+ func (mr *MockStateStoreMockRecorder) RegisterClient(ctx, clientID interface{}) *gomock.Call

+ func (mr *MockStateStoreMockRecorder) UpdateCheckPoints(ctx, checkpoints interface{}) *gomock.Call

+ func (mr *MockStateStoreMockRecorder) UpdateShards(ctx, stream, shards interface{}) *gomock.Call

type Record

+ ShardID string

+ type ShardCheckPoint struct

+ SequenceNumber string

+ ShardID string

+ Stream string

+ UpdatedAt time.Time

type StateStore

+ UpdateCheckPoints func(ctx context.Context, checkpoints []*ShardCheckPoint) error

Jan 21, 2022

Changes in this version

type Config

+ ClientID string

Dec 16, 2021

Changes in this version

type Config

+ EFOMode bool

+ Region string

Dec 6, 2021

Changes in this version

type Config

+ StateStore *StateStore

+ type StateStore interface

+ DeregisterClient func(ctx context.Context, clientID string) error

+ GetShards func(ctx context.Context, stream string) (Shards, error)

+ ListAllAliveClientIDs func(ctx context.Context) ([]string, error)

+ ListCheckPoints func(ctx context.Context, stream string, shardIDs []string) (map[string]string, error)

+ PingClientAliveness func(ctx context.Context, clientID string) error

+ PruneClients func(ctx context.Context) error

+ RegisterClient func(ctx context.Context, clientID string) error

+ UpdateCheckPoint func(ctx context.Context, stream, shardID, seq string) error

+ UpdateShards func(ctx context.Context, stream string, shards Shards) error

Nov 19, 2021

Changes in this version

type Config

+ ScanInterval time.Duration

Oct 1, 2021

Changes in this version

+ type Shard struct

+ Closed bool

+ ID string

+ type Shards []*Shard

Sep 16, 2021

Changes in this version

+ var ErrEmptyShardIDs = errors.New("kinesumer: empty shard ids given")

+ var ErrNoShardCache = errors.New("kinesumer: shard cache not found")

+ type Config struct

+ App string

+ DynamoDBEndpoint string

+ DynamoDBRegion string

+ DynamoDBTable string

+ KinesisEndpoint string

+ KinesisRegion string

+ RoleARN string

+ ScanLimit int64

+ ScanTimeout time.Duration

+ type Kinesumer struct

+ func NewKinesumer(cfg *Config) (*Kinesumer, error)

+ func (k *Kinesumer) Close()

+ func (k *Kinesumer) Consume(streams []string) (<-chan *Record, error)

+ func (k *Kinesumer) Errors() <-chan error

+ func (k *Kinesumer) Refresh(streams []string)

+ type Record struct

+ Stream string