Created IsLessThan method to SequencePair to make code more readable

batchconsumer/batcher/message_batcher.go

@@ -13,6 +13,25 @@ type SequencePair struct {
 	SubSequence int
 }
 
+func (s SequencePair) IsLessThan(pair SequencePair) bool {
+	if s.Sequence == nil {
+		return false
+	}
+	if pair.Sequence == nil {
+		return true
+	}
+
+	cmp := s.Sequence.Cmp(pair.Sequence)
+	if cmp == -1 {
+		return true
+	}
+	if cmp == 1 {
+		return false
+	}
+
+	return s.SubSequence < pair.SubSequence
+}
+
 // Sync is used to allow a writer to syncronize with the batcher.
 // The writer declares how to write messages (via its `SendBatch` method), while the batcher
 // keeps track of messages written
@@ -110,11 +129,7 @@ func (b *batcher) updateSequenceNumbers(pair SequencePair) {
 	b.mux.Lock()
 	defer b.mux.Unlock()
 
-	isSmaller := b.smallestSeq.Sequence == nil ||
+	if pair.IsLessThan(b.smallestSeq) {
-		pair.Sequence.Cmp(b.smallestSeq.Sequence) == -1 ||
-		(pair.Sequence.Cmp(b.smallestSeq.Sequence) == 0 &&
-			pair.SubSequence < b.smallestSeq.SubSequence)
-	if isSmaller {
 		b.smallestSeq = SequencePair{pair.Sequence, pair.SubSequence}
 	}
 }

batchconsumer/writer.go

@@ -207,10 +207,7 @@ func (b *batchedWriter) CheckPointBatch(tag string) {
 			continue
 		}
 
-		isSmaller := smallest.Sequence == nil || // smallest.Sequence means batch just flushed
+		if pair.IsLessThan(smallest) {
-			pair.Sequence.Cmp(smallest.Sequence) == -1 ||
-			(pair.Sequence.Cmp(smallest.Sequence) == 0 && pair.SubSequence < smallest.SubSequence)
-		if isSmaller {
 			smallest = pair
 		}
 	}