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

Renamed IsEmpty to IsNil on SequencePair

Browse source
This commit is contained in:
Xavi Ramirez 2017-08-10 20:16:41 +00:00
parent 45fad863d0
commit 55aeecddd7
7 changed files with 14 additions and 14 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
batchconsumer/batcher.go
View file

@ -38,7 +38,7 @@ func (b *batcher) AddMessage(msg []byte, pair kcl.SequencePair) error {
} }
b.Batch = append(b.Batch, msg) b.Batch = append(b.Batch, msg)
if b.SmallestSeq.IsEmpty() || pair.IsLessThan(b.SmallestSeq) { if b.SmallestSeq.IsNil() || pair.IsLessThan(b.SmallestSeq) {
b.SmallestSeq = pair b.SmallestSeq = pair
} }
b.LastUpdated = time.Now() b.LastUpdated = time.Now()

4
batchconsumer/batchermanager.go
View file

@ -128,12 +128,12 @@ func (b *batcherManager) sendCheckpoint(
} }
// Check for empty because it's possible that no messages have been ignored // Check for empty because it's possible that no messages have been ignored
if smallest.IsEmpty() || batcher.SmallestSeq.IsLessThan(smallest) { if smallest.IsNil() || batcher.SmallestSeq.IsLessThan(smallest) {
smallest = batcher.SmallestSeq smallest = batcher.SmallestSeq
} }
} }
if !smallest.IsEmpty() { if !smallest.IsNil() {
b.chkpntManager.Checkpoint(smallest) b.chkpntManager.Checkpoint(smallest)
} }
} }

2
batchconsumer/checkpointmanager.go
View file

@ -73,7 +73,7 @@ func (cm *checkpointManager) startCheckpointHandler(
} }
} }
if !pair.IsEmpty() { if !pair.IsNil() {
checkpointer.Checkpoint(pair) checkpointer.Checkpoint(pair)
lastCheckpoint = time.Now() lastCheckpoint = time.Now()
stats.Counter("checkpoints-sent", 1) stats.Counter("checkpoints-sent", 1)

2
batchconsumer/writer.go
View file

@ -82,7 +82,7 @@ func (b *batchedWriter) ProcessRecords(records []kcl.Record) error {
} }
pair = kcl.SequencePair{seq, record.SubSequenceNumber} pair = kcl.SequencePair{seq, record.SubSequenceNumber}
if prevPair.IsEmpty() { // Handles on-start edge case where b.lastProcessSeq is empty if prevPair.IsNil() { // Handles on-start edge case where b.lastProcessSeq is empty
prevPair = pair prevPair = pair
} }

6
kcl/kcl.go
View file

@ -156,7 +156,7 @@ func (kclp *KCLProcess) Checkpoint(pair SequencePair) {
kclp.ckpmux.Lock() kclp.ckpmux.Lock()
defer kclp.ckpmux.Unlock() defer kclp.ckpmux.Unlock()
if kclp.nextCheckpointPair.IsEmpty() || kclp.nextCheckpointPair.IsLessThan(pair) { if kclp.nextCheckpointPair.IsNil() || kclp.nextCheckpointPair.IsLessThan(pair) {
kclp.nextCheckpointPair = pair kclp.nextCheckpointPair = pair
} }
} }
@ -189,7 +189,7 @@ func (kclp *KCLProcess) handleCheckpointAction(action ActionCheckpoint) error {
subSeq := action.SubSequenceNumber subSeq := action.SubSequenceNumber
kclp.ckpmux.Lock() kclp.ckpmux.Lock()
if !kclp.nextCheckpointPair.IsEmpty() { if !kclp.nextCheckpointPair.IsNil() {
tmp := kclp.nextCheckpointPair.Sequence.String() tmp := kclp.nextCheckpointPair.Sequence.String()
seq = &tmp seq = &tmp
subSeq = &kclp.nextCheckpointPair.SubSequence subSeq = &kclp.nextCheckpointPair.SubSequence
@ -279,7 +279,7 @@ func (kclp *KCLProcess) Run() {
} }
kclp.ckpmux.Lock() kclp.ckpmux.Lock()
if !kclp.nextCheckpointPair.IsEmpty() { if !kclp.nextCheckpointPair.IsNil() {
seq := kclp.nextCheckpointPair.Sequence.String() seq := kclp.nextCheckpointPair.Sequence.String()
subSeq := kclp.nextCheckpointPair.SubSequence subSeq := kclp.nextCheckpointPair.SubSequence

4
kcl/sequencepair.go
View file

@ -10,12 +10,12 @@ type SequencePair struct {
SubSequence int SubSequence int
} }
func (s SequencePair) IsEmpty() bool { func (s SequencePair) IsNil() bool {
return s.Sequence == nil return s.Sequence == nil
} }
func (s SequencePair) IsLessThan(pair SequencePair) bool { func (s SequencePair) IsLessThan(pair SequencePair) bool {
if s.IsEmpty() || pair.IsEmpty() { // empty pairs are incomparable if s.IsNil() || pair.IsNil() { // empty pairs are incomparable
return false return false
} }

8
kcl/sequencepair_test.go
View file

@ -46,9 +46,9 @@ func TestSequencePairIsLessThan(t *testing.T) {
func TestSequencePairEmpty(t *testing.T) { func TestSequencePairEmpty(t *testing.T) {
assert := assert.New(t) assert := assert.New(t)
assert.True(SequencePair{nil, 0}.IsEmpty()) assert.True(SequencePair{nil, 0}.IsNil())
assert.True(SequencePair{nil, 10000}.IsEmpty()) assert.True(SequencePair{nil, 10000}.IsNil())
assert.False(SequencePair{big.NewInt(10), 0}.IsEmpty()) assert.False(SequencePair{big.NewInt(10), 0}.IsNil())
assert.False(SequencePair{big.NewInt(0), 1000}.IsEmpty()) assert.False(SequencePair{big.NewInt(0), 1000}.IsNil())
} }