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

Breaking change to KCL. Created Checkpointer interface to make testing easier.

Browse source
This commit is contained in:
Xavi Ramirez 2017-07-21 01:04:34 +00:00
parent 164b9923be
commit b0f769bfa7
2 changed files with 16 additions and 10 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
cmd/consumer/main.go
View file

@ -10,7 +10,7 @@ import (
) )
type sampleRecordProcessor struct { type sampleRecordProcessor struct {
checkpointer *kcl.Checkpointer checkpointer kcl.Checkpointer
checkpointRetries int checkpointRetries int
checkpointFreq time.Duration checkpointFreq time.Duration
largestSeq *big.Int largestSeq *big.Int
@ -25,7 +25,7 @@ func newSampleRecordProcessor() *sampleRecordProcessor {
} }
} }
func (srp *sampleRecordProcessor) Initialize(shardID string, checkpointer *kcl.Checkpointer) error { func (srp *sampleRecordProcessor) Initialize(shardID string, checkpointer kcl.Checkpointer) error {
srp.lastCheckpoint = time.Now() srp.lastCheckpoint = time.Now()
srp.checkpointer = checkpointer srp.checkpointer = checkpointer
return nil return nil

22
kcl/kcl.go
View file

@ -12,11 +12,17 @@ import (
) )
type RecordProcessor interface { type RecordProcessor interface {
Initialize(shardID string, checkpointer *Checkpointer) error Initialize(shardID string, checkpointer Checkpointer) error
ProcessRecords(records []Record) error ProcessRecords(records []Record) error
Shutdown(reason string) error Shutdown(reason string) error
} }
type Checkpointer interface {
Checkpoint(sequenceNumber *string, subSequenceNumber *int) error
CheckpointWithRetry(sequenceNumber *string, subSequenceNumber *int, retryCount int) error
Shutdown()
}
type CheckpointError struct { type CheckpointError struct {
e string e string
} }
@ -25,13 +31,13 @@ func (ce CheckpointError) Error() string {
return ce.e return ce.e
} }
type Checkpointer struct { type checkpointer struct {
mux sync.Mutex mux sync.Mutex
ioHandler ioHandler ioHandler ioHandler
} }
func (c *Checkpointer) getAction() (interface{}, error) { func (c *checkpointer) getAction() (interface{}, error) {
line, err := c.ioHandler.readLine() line, err := c.ioHandler.readLine()
if err != nil { if err != nil {
return nil, err return nil, err
@ -43,7 +49,7 @@ func (c *Checkpointer) getAction() (interface{}, error) {
return action, nil return action, nil
} }
func (c *Checkpointer) Checkpoint(sequenceNumber *string, subSequenceNumber *int) error { func (c *checkpointer) Checkpoint(sequenceNumber *string, subSequenceNumber *int) error {
c.mux.Lock() c.mux.Lock()
defer c.mux.Unlock() defer c.mux.Unlock()
@ -74,7 +80,7 @@ func (c *Checkpointer) Checkpoint(sequenceNumber *string, subSequenceNumber *int
// CheckpointWithRetry tries to save a checkPoint up to `retryCount` + 1 times. // CheckpointWithRetry tries to save a checkPoint up to `retryCount` + 1 times.
// `retryCount` should be >= 0 // `retryCount` should be >= 0
func (c *Checkpointer) CheckpointWithRetry( func (c *checkpointer) CheckpointWithRetry(
sequenceNumber *string, subSequenceNumber *int, retryCount int, sequenceNumber *string, subSequenceNumber *int, retryCount int,
) error { ) error {
sleepDuration := 5 * time.Second sleepDuration := 5 * time.Second
@ -108,7 +114,7 @@ func (c *Checkpointer) CheckpointWithRetry(
return nil return nil
} }
func (c *Checkpointer) Shutdown() { func (c *checkpointer) Shutdown() {
c.CheckpointWithRetry(nil, nil, 5) c.CheckpointWithRetry(nil, nil, 5)
} }
@ -225,7 +231,7 @@ func New(inputFile io.Reader, outputFile, errorFile io.Writer, recordProcessor R
} }
return &KCLProcess{ return &KCLProcess{
ioHandler: i, ioHandler: i,
checkpointer: &Checkpointer{ checkpointer: &checkpointer{
ioHandler: i, ioHandler: i,
}, },
recordProcessor: recordProcessor, recordProcessor: recordProcessor,
@ -234,7 +240,7 @@ func New(inputFile io.Reader, outputFile, errorFile io.Writer, recordProcessor R
type KCLProcess struct { type KCLProcess struct {
ioHandler ioHandler ioHandler ioHandler
checkpointer *Checkpointer checkpointer Checkpointer
recordProcessor RecordProcessor recordProcessor RecordProcessor
} }