unblock blocked event emitters when stopped (#790)

utils/event_emitter.go

@@ -134,8 +134,8 @@ type eventObserverListObserver[V any] struct {
 }
 
 func (o *eventObserverListObserver[V]) Stop() {
-	o.l.stopObserving(o)
 	o.EventObserver.Stop()
+	o.l.stopObserving(o)
 }
 
 func (l *EventObserverList[V]) Observe() EventObserver[V] {

utils/event_emitter_test.go

@@ -17,6 +17,7 @@ package utils
 import (
 	"sort"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/require"
 )
@@ -60,6 +61,78 @@ func TestEventEmitter(t *testing.T) {
 		o.Stop()
 		require.True(t, closeCalled)
 	})
+
+	t.Run("stop unblocks blocking observers", func(t *testing.T) {
+		observer, emit := NewEventObserver[int](func() {})
+
+		list := NewEventObserverList[int](EventEmitterParams{
+			QueueSize: DefaultEventQueueSize,
+			Blocking:  true,
+		})
+
+		emitter := NewEventEmitter[int, int](EventEmitterParams{
+			QueueSize: DefaultEventQueueSize,
+			Blocking:  true,
+		})
+
+		cases := []struct {
+			label    string
+			emit     func()
+			observer EventObserver[int]
+		}{
+			{
+				label:    "observer",
+				emit:     func() { emit(0) },
+				observer: observer,
+			},
+			{
+				label:    "list",
+				emit:     func() { list.Emit(0) },
+				observer: list.Observe(),
+			},
+			{
+				label:    "emitter",
+				emit:     func() { emitter.Emit(0, 0) },
+				observer: emitter.Observe(0),
+			},
+		}
+
+		for _, c := range cases {
+			t.Run(c.label, func(t *testing.T) {
+				emitDone := make(chan struct{})
+				stopDone := make(chan struct{})
+				ready := make(chan struct{})
+
+				go func() {
+					for i := 0; i < DefaultEventQueueSize; i++ {
+						c.emit()
+					}
+					close(ready)
+					c.emit()
+					close(emitDone)
+				}()
+
+				go func() {
+					<-ready
+					time.Sleep(100 * time.Millisecond)
+					c.observer.Stop()
+					close(stopDone)
+				}()
+
+				select {
+				case <-emitDone:
+				case <-time.After(time.Second):
+					require.FailNow(t, "timeout waiting for emit to unblock")
+				}
+
+				select {
+				case <-stopDone:
+				case <-time.After(time.Second):
+					require.FailNow(t, "timeout waiting for stop to unblock")
+				}
+			})
+		}
+	})
 }
 
 func BenchmarkEventEmitter(b *testing.B) {