fix: avoid duplicate sequence numbers and fix late subscriptions (#2)

eventbus.go

@@ -41,6 +41,7 @@ func (b *Bus[T]) subscribe(topic string, once bool) (<-chan T, func()) {
 
 	if _, ok := b.subscriptions[topic]; !ok {
 		b.subscriptions[topic] = make([]subscription[T], 0)
+		b.publishSequences[topic] = 1
 	}
 
 	sub := subscription[T]{
@@ -50,7 +51,7 @@ func (b *Bus[T]) subscribe(topic string, once bool) (<-chan T, func()) {
 	}
 	// once subscriptions don't need serializer
 	if !once {
-		sub.serializer = NewSerializer()
+		sub.serializer = NewSerializer(b.publishSequences[topic] - 1)
 	}
 
 	b.subscriptions[topic] = append(b.subscriptions[topic], sub)

eventbus_test.go

@@ -24,7 +24,7 @@ func TestBus_Subscribe(t *testing.T) {
 	cA1, uA1 := bus.Subscribe(topicA)
 	cA2, _ := bus.Subscribe(topicA)
 	cB1, _ := bus.Subscribe(topicB)
-	a1Counter, a2Counter, b1Counter := 0, 0, 0
+	a1Counter, a2Counter, b1Counter, b2Counter := 0, 0, 0, 0
 
 	ctx, ctxCancel := context.WithCancel(context.Background())
 	defer ctxCancel()
@@ -54,11 +54,25 @@ func TestBus_Subscribe(t *testing.T) {
 	bus.Publish(topicA, eventsA[1])
 	bus.Publish(topicB, eventsB[0])
 
+	cB2, _ := bus.Subscribe(topicB)
+	go func() {
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-cB2:
+				b2Counter++
+			}
+		}
+	}()
+	bus.Publish(topicB, eventsB[1])
+
 	time.Sleep(1 * time.Second)
 
 	assert.Equal(t, 1, a1Counter)
 	assert.Equal(t, 2, a2Counter)
-	assert.Equal(t, 1, b1Counter)
+	assert.Equal(t, 2, b1Counter)
+	assert.Equal(t, 1, b2Counter)
 
 	assert.Equal(t, 2, len(bus.subscriptions))
 	assert.Equal(t, 2, len(bus.publishSequences))

serializer.go

@@ -47,7 +47,13 @@ func (s *Serializer) Execute(cb callback, sequence uint64) {
 
 		item := s.queue[0]
 
-		if item.sequence != s.lastExecutedSequence && item.sequence != s.lastExecutedSequence+1 {
+		if item.sequence <= s.lastExecutedSequence {
+			s.queue = s.queue[1:]
+
+			continue
+		}
+
+		if item.sequence != s.lastExecutedSequence+1 {
 			break
 		}
 
@@ -58,8 +64,9 @@ func (s *Serializer) Execute(cb callback, sequence uint64) {
 }
 
 // NewSerializer creates a new serializer.
-func NewSerializer() *Serializer {
+func NewSerializer(lastExecutedSequence uint64) *Serializer {
 	return &Serializer{
-		queue: make([]queueItem, 0),
+		queue:                make([]queueItem, 0),
+		lastExecutedSequence: lastExecutedSequence,
 	}
 }

serializer_test.go

@@ -7,7 +7,7 @@ import (
 )
 
 func TestSerializer_Execute(t *testing.T) {
-	serializer := eventbus.NewSerializer()
+	serializer := eventbus.NewSerializer(0)
 	b1 := false
 	b2 := false
 	b3 := false
@@ -16,7 +16,7 @@ func TestSerializer_Execute(t *testing.T) {
 
 	serializer.Execute(func() {
 		b1 = true
-	}, 0)
+	}, 1)
 	assert.Equal(t, true, b1)
 	assert.Equal(t, false, b2)
 	assert.Equal(t, false, b3)
@@ -25,7 +25,7 @@ func TestSerializer_Execute(t *testing.T) {
 
 	serializer.Execute(func() {
 		b3 = true
-	}, 2)
+	}, 3)
 	assert.Equal(t, true, b1)
 	assert.Equal(t, false, b2)
 	assert.Equal(t, false, b3)
@@ -34,7 +34,7 @@ func TestSerializer_Execute(t *testing.T) {
 
 	serializer.Execute(func() {
 		b4 = true
-	}, 2)
+	}, 3)
 	assert.Equal(t, true, b1)
 	assert.Equal(t, false, b2)
 	assert.Equal(t, false, b3)
@@ -43,7 +43,7 @@ func TestSerializer_Execute(t *testing.T) {
 
 	serializer.Execute(func() {
 		b5 = true
-	}, 3)
+	}, 4)
 	assert.Equal(t, true, b1)
 	assert.Equal(t, false, b2)
 	assert.Equal(t, false, b3)
@@ -52,10 +52,10 @@ func TestSerializer_Execute(t *testing.T) {
 
 	serializer.Execute(func() {
 		b2 = true
-	}, 1)
+	}, 2)
 	assert.Equal(t, true, b1)
 	assert.Equal(t, true, b2)
-	assert.Equal(t, true, b3)
+	assert.Equal(t, false, b3)
 	assert.Equal(t, true, b4)
 	assert.Equal(t, true, b5)
 }