From 292f255131c7908e48505b1ea954ac810791cd80 Mon Sep 17 00:00:00 2001 From: ppenguin Date: Fri, 23 Oct 2026 13:48:59 +0200 Subject: [PATCH 1/2] fix(service): keep subscriptions made before Start, drop publishes before it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Services.Start runs EventServiceWS/ActionServiceWS.Start in goroutines; YSK subscribes right after → Start's subscriberChannels reset dropped that subscription (cards stop updating, depending on scheduling) - Publish/Trigger before Start: select on (*s.ctx).Done() with nil ctx → panic - Start keeps an existing subscriber map; Publish/Trigger drop (and log) until Start ran - tests: subscription before Start receives; publish before Start returns Co-Authored-By: Claude Opus 5.5 --- pkg/ysk/ysk_test.go | 5 ++-- service/action_service_websocket.go | 9 ++++-- service/event_service_websocket.go | 20 +++++++++---- service/event_service_websocket_test.go | 37 +++++++++++++++++++++++++ service/event_type_service_test.go | 27 ++++++++++++++---- 5 files changed, 83 insertions(+), 15 deletions(-) diff --git a/pkg/ysk/ysk_test.go b/pkg/ysk/ysk_test.go index e359995..13866d7 100644 --- a/pkg/ysk/ysk_test.go +++ b/pkg/ysk/ysk_test.go @@ -25,9 +25,8 @@ func setup(t *testing.T) (*service.EventServiceWS, *service.YSKService, func()) ctx := context.Background() go s.Start(&ctx) - // EventServiceWS exposes no readiness signal and Publish dereferences the - // context it installs in Start, so the boot wait cannot be made conditional. - // Subscribing before Start would also be wiped by its subscriberChannels reset. + // EventServiceWS exposes no readiness signal and Publish drops events until + // Start has run, so the boot wait cannot be made conditional. time.Sleep(1 * time.Second) return wsService, yskService, func() { diff --git a/service/action_service_websocket.go b/service/action_service_websocket.go index f934329..55c2e32 100644 --- a/service/action_service_websocket.go +++ b/service/action_service_websocket.go @@ -23,8 +23,10 @@ type ActionServiceWS struct { } func (s *ActionServiceWS) Trigger(action model.Action) { - if s.inboundChannel == nil { + // before Start: drop (the select below dereferences ctx) + if s.inboundChannel == nil || s.ctx == nil { logger.Error("error when triggering action via websocket", zap.Error(ErrInboundChannelNotFound)) + return } if action.Timestamp == 0 { @@ -122,7 +124,10 @@ func (s *ActionServiceWS) Start(ctx *context.Context) { s.mutex = sync.Mutex{} s.inboundChannel = make(chan model.Action) - s.subscriberChannels = make(map[string]map[string][]chan model.Action) + // keep subscriptions made before Start (see EventServiceWS.Start) + if s.subscriberChannels == nil { + s.subscriberChannels = make(map[string]map[string][]chan model.Action) + } s.stop = make(chan struct{}) defer func() { diff --git a/service/event_service_websocket.go b/service/event_service_websocket.go index 0c0d913..71ab747 100644 --- a/service/event_service_websocket.go +++ b/service/event_service_websocket.go @@ -30,8 +30,15 @@ type EventServiceWS struct { var mutex = &sync.Mutex{} func (s *EventServiceWS) Publish(event model.Event) { - if s.inboundChannel == nil { + // Start runs in a goroutine, sets both under the lock; before that: drop (no nil-ctx panic) + inboundChannel, ctx := func() (chan model.Event, *context.Context) { + mutex.Lock() + defer mutex.Unlock() + return s.inboundChannel, s.ctx + }() + if inboundChannel == nil || ctx == nil { logger.Error("error when publishing event via websocket", zap.Error(ErrInboundChannelNotFound)) + return } if event.Timestamp == 0 { @@ -44,10 +51,10 @@ func (s *EventServiceWS) Publish(event model.Event) { // goroutine of its own, and an event dropped here because the dispatch was // busy with the one before was an alert never sent. select { - case s.inboundChannel <- event: + case inboundChannel <- event: - case <-(*s.ctx).Done(): - if err := (*s.ctx).Err(); err != nil { + case <-(*ctx).Done(): + if err := (*ctx).Err(); err != nil { logger.Info(err.Error()) } return @@ -141,7 +148,10 @@ func (s *EventServiceWS) Start(ctx *context.Context) { s.ctx = ctx s.inboundChannel = make(chan model.Event) - s.subscriberChannels = make(map[string]map[string][]chan model.Event) + // keep subscriptions made before Start (YSK subscribes right after Services.Start) + if s.subscriberChannels == nil { + s.subscriberChannels = make(map[string]map[string][]chan model.Event) + } s.stop = make(chan struct{}) }() diff --git a/service/event_service_websocket_test.go b/service/event_service_websocket_test.go index 7c29a39..4a5f4e0 100644 --- a/service/event_service_websocket_test.go +++ b/service/event_service_websocket_test.go @@ -102,3 +102,40 @@ func TestEventsPublishedTogetherAllArrive(t *testing.T) { } } } + +// A subscription made before Start (YSK subscribes right after Services.Start, +// which starts the dispatch in a goroutine) must survive Start. +func TestSubscriptionBeforeStartSurvivesStart(t *testing.T) { + ws := newEventService(t) + + channel, err := ws.Subscribe("app-management", []string{"backup:error"}) + assert.NilError(t, err) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go ws.Start(&ctx) + + // Publish drops events until the dispatch runs: publish again until one arrives. + deadline := time.After(10 * time.Second) + tick := time.NewTicker(10 * time.Millisecond) + defer tick.Stop() + for { + select { + case event := <-channel: + assert.Equal(t, event.Name, "backup:error") + return + case <-tick.C: + ws.Publish(model.Event{SourceID: "app-management", Name: "backup:error"}) + case <-deadline: + t.Fatal("a subscription made before Start never received an event") + } + } +} + +// A Publish before Start is dropped; it used to dereference the nil context. +func TestPublishBeforeStartIsDropped(t *testing.T) { + ws := newEventService(t) + within(t, "Publish before Start", func() { + ws.Publish(model.Event{SourceID: "app-management", Name: "backup:error"}) + }) +} diff --git a/service/event_type_service_test.go b/service/event_type_service_test.go index 658ef8e..224a0a4 100644 --- a/service/event_type_service_test.go +++ b/service/event_type_service_test.go @@ -3,7 +3,9 @@ package service import ( "context" "testing" + "time" + "github.com/ReCasaOS/CasaOS-Common/utils/logger" "github.com/ReCasaOS/CasaOS-MessageBus/model" "github.com/ReCasaOS/CasaOS-MessageBus/repository" "go.uber.org/goleak" @@ -11,6 +13,7 @@ import ( ) func TestEventTypeService(t *testing.T) { + logger.LogInitConsoleOnly() // Publish logs while the loop is not up yet defer goleak.VerifyNone(t) // new repository @@ -90,11 +93,25 @@ func TestEventTypeService(t *testing.T) { // }, } - wsService.Publish(expectedEvent) - assert.NilError(t, err) - - actualEvent, ok := <-outputChannel - assert.Equal(t, ok, true) + // Start runs in a goroutine; Publish drops until the loop runs → republish + actualEvent := publishUntilReceived(t, wsService, expectedEvent, outputChannel) assert.DeepEqual(t, model.Event{SourceID: actualEvent.SourceID, Name: actualEvent.Name, Properties: actualEvent.Properties}, expectedEvent) } } + +func publishUntilReceived(t *testing.T, wsService *EventServiceWS, event model.Event, received <-chan model.Event) model.Event { + t.Helper() + deadline := time.After(10 * time.Second) + tick := time.NewTicker(10 * time.Millisecond) + defer tick.Stop() + for { + wsService.Publish(event) + select { + case actual := <-received: + return actual + case <-tick.C: + case <-deadline: + t.Fatalf("event %s/%s never arrived", event.SourceID, event.Name) + } + } +} From f850e356c32a5f0fe4ab5c47398086824e7e4080 Mon Sep 17 00:00:00 2001 From: ppenguin Date: Sat, 24 Oct 2026 15:52:26 +0200 Subject: [PATCH 2/2] fix(service): action service locked like the event service; Ready() replaces republish loops MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - ActionServiceWS: Start no longer resets the mutex; Start setup/teardown, Trigger's read, Subscribe, dispatch, heartbeat under it → pre-Start subscriptions kept, no race with Start - Trigger waits for the dispatch like Publish (was: dropped when the dispatch was busy); routes trigger from their own goroutine - Ready(): closed by Start once its state is in → tests wait on it: no 10 ms republish loops (duplicates under -race, leaked forwarders), no 1 s sleep in pkg/ysk - before Start: Publish/Trigger drop with Warn - tests: action subscription before Start, Trigger before Start, Subscribe/Trigger concurrent with Start (-race) Co-Authored-By: Claude Opus 5.5 --- pkg/ysk/ysk_test.go | 4 +- service/action_service_websocket.go | 135 ++++++++++++++--------- service/action_service_websocket_test.go | 109 ++++++++++++++++++ service/event_service_websocket.go | 32 +++++- service/event_service_websocket_test.go | 23 ++-- service/event_type_service_test.go | 27 +---- 6 files changed, 236 insertions(+), 94 deletions(-) create mode 100644 service/action_service_websocket_test.go diff --git a/pkg/ysk/ysk_test.go b/pkg/ysk/ysk_test.go index 13866d7..d3621b7 100644 --- a/pkg/ysk/ysk_test.go +++ b/pkg/ysk/ysk_test.go @@ -25,9 +25,7 @@ func setup(t *testing.T) (*service.EventServiceWS, *service.YSKService, func()) ctx := context.Background() go s.Start(&ctx) - // EventServiceWS exposes no readiness signal and Publish drops events until - // Start has run, so the boot wait cannot be made conditional. - time.Sleep(1 * time.Second) + <-wsService.Ready() return wsService, yskService, func() { repository.Close() diff --git a/service/action_service_websocket.go b/service/action_service_websocket.go index 55c2e32..ea6848a 100644 --- a/service/action_service_websocket.go +++ b/service/action_service_websocket.go @@ -14,18 +14,43 @@ import ( type ActionServiceWS struct { typeService *ActionTypeService - ctx *context.Context + // mutex guards every field below: Start runs in a goroutine of its own, + // Trigger and Subscribe come from the routes mutex sync.Mutex + ctx *context.Context stop chan struct{} inboundChannel chan model.Action subscriberChannels map[string]map[string][]chan model.Action + + // ready: closed once Start has installed its state; Trigger delivers from then on + ready chan struct{} + readyOnce sync.Once +} + +// Ready is closed once Start runs: actions triggered from then on are delivered. +func (s *ActionServiceWS) Ready() <-chan struct{} { + s.mutex.Lock() + defer s.mutex.Unlock() + return s.readyLocked() +} + +func (s *ActionServiceWS) readyLocked() chan struct{} { + if s.ready == nil { + s.ready = make(chan struct{}) + } + return s.ready } func (s *ActionServiceWS) Trigger(action model.Action) { - // before Start: drop (the select below dereferences ctx) - if s.inboundChannel == nil || s.ctx == nil { - logger.Error("error when triggering action via websocket", zap.Error(ErrInboundChannelNotFound)) + // Start runs in a goroutine, sets both under the lock; before that: drop (no nil-ctx panic) + inboundChannel, ctx := func() (chan model.Action, *context.Context) { + s.mutex.Lock() + defer s.mutex.Unlock() + return s.inboundChannel, s.ctx + }() + if inboundChannel == nil || ctx == nil { + logger.Warn("action triggered before the bus started: dropped", zap.String("sourceID", action.SourceID), zap.String("name", action.Name)) return } @@ -35,16 +60,17 @@ func (s *ActionServiceWS) Trigger(action model.Action) { // TODO - ensure properties are valid for action type + // Waits for the dispatch to take it, never long (it waits for no + // subscriber), as EventServiceWS.Publish does: the route triggers from a + // goroutine of its own. select { - case s.inboundChannel <- action: + case inboundChannel <- action: - case <-(*s.ctx).Done(): - if err := (*s.ctx).Err(); err != nil { + case <-(*ctx).Done(): + if err := (*ctx).Err(); err != nil { logger.Info(err.Error()) } return - - default: // drop action if no one is listening } } @@ -71,6 +97,9 @@ func (s *ActionServiceWS) Subscribe(sourceID string, names []string) (chan model } } + s.mutex.Lock() + defer s.mutex.Unlock() + if s.subscriberChannels == nil { s.subscriberChannels = make(map[string]map[string][]chan model.Action) } @@ -120,17 +149,27 @@ func (s *ActionServiceWS) Unsubscribe(sourceID string, name string, c chan model } func (s *ActionServiceWS) Start(ctx *context.Context) { - s.ctx = ctx - s.mutex = sync.Mutex{} + inboundChannel := func() chan model.Action { + s.mutex.Lock() + defer s.mutex.Unlock() - s.inboundChannel = make(chan model.Action) - // keep subscriptions made before Start (see EventServiceWS.Start) - if s.subscriberChannels == nil { - s.subscriberChannels = make(map[string]map[string][]chan model.Action) - } - s.stop = make(chan struct{}) + s.ctx = ctx + + s.inboundChannel = make(chan model.Action) + // keep subscriptions made before Start (see EventServiceWS.Start) + if s.subscriberChannels == nil { + s.subscriberChannels = make(map[string]map[string][]chan model.Action) + } + s.stop = make(chan struct{}) + ready := s.readyLocked() + s.readyOnce.Do(func() { close(ready) }) + return s.inboundChannel + }() defer func() { + s.mutex.Lock() + defer s.mutex.Unlock() + if s.subscriberChannels != nil { for sourceID, source := range s.subscriberChannels { for actionName, subscribers := range source { @@ -164,60 +203,54 @@ func (s *ActionServiceWS) Start(ctx *context.Context) { for { select { - case <-(*s.ctx).Done(): + case <-(*ctx).Done(): return - case action, ok := <-s.inboundChannel: + case action, ok := <-inboundChannel: if !ok { return } - if s.subscriberChannels == nil { - continue - } - - if s.subscriberChannels[action.SourceID] == nil { - continue - } - - if s.subscriberChannels[action.SourceID][action.Name] == nil { - continue - } + // under the lock Subscribe and Unsubscribe change the lists with (see + // EventServiceWS.Start) + func() { + s.mutex.Lock() + defer s.mutex.Unlock() - for _, c := range s.subscriberChannels[action.SourceID][action.Name] { - select { - case c <- action: - case <-(*s.ctx).Done(): + if s.subscriberChannels == nil || s.subscriberChannels[action.SourceID] == nil { return - default: // drop event if no one is listening - continue } - } - case <-ticker.C: - if s.subscriberChannels == nil { - continue - } + for _, c := range s.subscriberChannels[action.SourceID][action.Name] { + select { + case c <- action: + default: // drop action if no one is listening + } + } + }() + case <-ticker.C: heartbeat := model.Action{ SourceID: common.MessageBusSourceID, Name: common.MessageBusHeartbeatName, Timestamp: time.Now().Unix(), } - for _, source := range s.subscriberChannels { - for _, subscribers := range source { - for _, subscriber := range subscribers { - select { - case subscriber <- heartbeat: - case <-(*s.ctx).Done(): - return - default: // drop event if no one is listening - continue + func() { + s.mutex.Lock() + defer s.mutex.Unlock() + + for _, source := range s.subscriberChannels { + for _, subscribers := range source { + for _, subscriber := range subscribers { + select { + case subscriber <- heartbeat: + default: // drop event if no one is listening + } } } } - } + }() } } } diff --git a/service/action_service_websocket_test.go b/service/action_service_websocket_test.go new file mode 100644 index 0000000..20ff6f5 --- /dev/null +++ b/service/action_service_websocket_test.go @@ -0,0 +1,109 @@ +package service + +import ( + "context" + "sync" + "testing" + "time" + + "github.com/ReCasaOS/CasaOS-Common/utils/logger" + "github.com/ReCasaOS/CasaOS-MessageBus/model" + "github.com/ReCasaOS/CasaOS-MessageBus/repository" + "go.uber.org/goleak" + "gotest.tools/assert" +) + +func newActionService(t *testing.T) *ActionServiceWS { + t.Helper() + logger.LogInitConsoleOnly() + repository, err := repository.NewDatabaseRepositoryInMemory() + assert.NilError(t, err) + t.Cleanup(func() { repository.Close() }) + typeService := NewActionTypeService(&repository) + _, err = typeService.RegisterActionType(model.ActionType{SourceID: "app-management", Name: "app:install"}) + assert.NilError(t, err) + return NewActionServiceWS(typeService) +} + +func receiveAction(t *testing.T, channel chan model.Action) model.Action { + t.Helper() + select { + case action := <-channel: + return action + case <-time.After(5 * time.Second): + t.Fatal("no action arrived") + return model.Action{} + } +} + +// A subscription made before Start must survive Start (see the event service). +func TestActionSubscriptionBeforeStartSurvivesStart(t *testing.T) { + // a cleanup: runs after the repository's (registered later, run first) + t.Cleanup(func() { goleak.VerifyNone(t) }) + ws := newActionService(t) + + channel, err := ws.Subscribe("app-management", []string{"app:install"}) + assert.NilError(t, err) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go ws.Start(&ctx) + <-ws.Ready() + + within(t, "Trigger after Start", func() { + ws.Trigger(model.Action{SourceID: "app-management", Name: "app:install"}) + }) + assert.Equal(t, receiveAction(t, channel).Name, "app:install") +} + +// A Trigger before Start is dropped; it used to dereference the nil context. +func TestTriggerBeforeStartIsDropped(t *testing.T) { + ws := newActionService(t) + within(t, "Trigger before Start", func() { + ws.Trigger(model.Action{SourceID: "app-management", Name: "app:install"}) + }) +} + +// Subscribe and Trigger from the routes while Start installs its state, and +// while it dispatches: -race finds an unguarded field. +func TestActionSubscribeAndTriggerRaceWithStart(t *testing.T) { + // a cleanup: runs after the repository's (registered later, run first) + t.Cleanup(func() { goleak.VerifyNone(t) }) + ws := newActionService(t) + + ctx, cancel := context.WithCancel(context.Background()) + stopped := make(chan struct{}) + + var wg sync.WaitGroup + channels := make(chan chan model.Action, 8) + for range 8 { + wg.Add(2) + go func() { + defer wg.Done() + channel, err := ws.Subscribe("app-management", []string{"app:install"}) + assert.Check(t, err) + channels <- channel + }() + go func() { + defer wg.Done() + ws.Trigger(model.Action{SourceID: "app-management", Name: "app:install"}) + }() + } + go func() { + ws.Start(&ctx) + close(stopped) + }() + wg.Wait() + close(channels) + + // every subscription made concurrently with Start is kept + <-ws.Ready() + ws.Trigger(model.Action{SourceID: "app-management", Name: "app:install"}) + for channel := range channels { + // buffered (1): the last action, or an earlier one if it was still waiting + assert.Equal(t, receiveAction(t, channel).Name, "app:install") + } + + cancel() + <-stopped +} diff --git a/service/event_service_websocket.go b/service/event_service_websocket.go index 71ab747..e1cabd2 100644 --- a/service/event_service_websocket.go +++ b/service/event_service_websocket.go @@ -23,12 +23,30 @@ type EventServiceWS struct { ctx *context.Context stop chan struct{} + // ready: closed once Start has installed its state; Publish delivers from then on + ready chan struct{} + readyOnce sync.Once + inboundChannel chan model.Event subscriberChannels map[string]map[string][]chan model.Event } var mutex = &sync.Mutex{} +// Ready is closed once Start runs: events published from then on are delivered. +func (s *EventServiceWS) Ready() <-chan struct{} { + mutex.Lock() + defer mutex.Unlock() + return s.readyLocked() +} + +func (s *EventServiceWS) readyLocked() chan struct{} { + if s.ready == nil { + s.ready = make(chan struct{}) + } + return s.ready +} + func (s *EventServiceWS) Publish(event model.Event) { // Start runs in a goroutine, sets both under the lock; before that: drop (no nil-ctx panic) inboundChannel, ctx := func() (chan model.Event, *context.Context) { @@ -37,7 +55,7 @@ func (s *EventServiceWS) Publish(event model.Event) { return s.inboundChannel, s.ctx }() if inboundChannel == nil || ctx == nil { - logger.Error("error when publishing event via websocket", zap.Error(ErrInboundChannelNotFound)) + logger.Warn("event published before the bus started: dropped", zap.String("sourceID", event.SourceID), zap.String("name", event.Name)) return } @@ -141,7 +159,7 @@ func (s *EventServiceWS) Unsubscribe(sourceID string, name string, c chan model. } func (s *EventServiceWS) Start(ctx *context.Context) { - func() { + inboundChannel := func() chan model.Event { mutex.Lock() defer mutex.Unlock() @@ -153,9 +171,15 @@ func (s *EventServiceWS) Start(ctx *context.Context) { s.subscriberChannels = make(map[string]map[string][]chan model.Event) } s.stop = make(chan struct{}) + ready := s.readyLocked() + s.readyOnce.Do(func() { close(ready) }) + return s.inboundChannel }() defer func() { + mutex.Lock() + defer mutex.Unlock() + if s.subscriberChannels != nil { for sourceID, source := range s.subscriberChannels { for eventName, subscribers := range source { @@ -189,10 +213,10 @@ func (s *EventServiceWS) Start(ctx *context.Context) { for { select { - case <-(*s.ctx).Done(): + case <-(*ctx).Done(): return - case event, ok := <-s.inboundChannel: + case event, ok := <-inboundChannel: if !ok { return } diff --git a/service/event_service_websocket_test.go b/service/event_service_websocket_test.go index 4a5f4e0..d562fb7 100644 --- a/service/event_service_websocket_test.go +++ b/service/event_service_websocket_test.go @@ -114,21 +114,16 @@ func TestSubscriptionBeforeStartSurvivesStart(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() go ws.Start(&ctx) + <-ws.Ready() - // Publish drops events until the dispatch runs: publish again until one arrives. - deadline := time.After(10 * time.Second) - tick := time.NewTicker(10 * time.Millisecond) - defer tick.Stop() - for { - select { - case event := <-channel: - assert.Equal(t, event.Name, "backup:error") - return - case <-tick.C: - ws.Publish(model.Event{SourceID: "app-management", Name: "backup:error"}) - case <-deadline: - t.Fatal("a subscription made before Start never received an event") - } + within(t, "Publish after Start", func() { + ws.Publish(model.Event{SourceID: "app-management", Name: "backup:error"}) + }) + select { + case event := <-channel: + assert.Equal(t, event.Name, "backup:error") + case <-time.After(5 * time.Second): + t.Fatal("a subscription made before Start never received an event") } } diff --git a/service/event_type_service_test.go b/service/event_type_service_test.go index 224a0a4..95ba1e1 100644 --- a/service/event_type_service_test.go +++ b/service/event_type_service_test.go @@ -3,9 +3,7 @@ package service import ( "context" "testing" - "time" - "github.com/ReCasaOS/CasaOS-Common/utils/logger" "github.com/ReCasaOS/CasaOS-MessageBus/model" "github.com/ReCasaOS/CasaOS-MessageBus/repository" "go.uber.org/goleak" @@ -13,7 +11,6 @@ import ( ) func TestEventTypeService(t *testing.T) { - logger.LogInitConsoleOnly() // Publish logs while the loop is not up yet defer goleak.VerifyNone(t) // new repository @@ -30,6 +27,7 @@ func TestEventTypeService(t *testing.T) { defer cancel() go wsService.Start(&ctx) + <-wsService.Ready() sourceID := "Foo" eventNames := []string{"Bar", "Baz"} @@ -93,25 +91,10 @@ func TestEventTypeService(t *testing.T) { // }, } - // Start runs in a goroutine; Publish drops until the loop runs → republish - actualEvent := publishUntilReceived(t, wsService, expectedEvent, outputChannel) - assert.DeepEqual(t, model.Event{SourceID: actualEvent.SourceID, Name: actualEvent.Name, Properties: actualEvent.Properties}, expectedEvent) - } -} + wsService.Publish(expectedEvent) -func publishUntilReceived(t *testing.T, wsService *EventServiceWS, event model.Event, received <-chan model.Event) model.Event { - t.Helper() - deadline := time.After(10 * time.Second) - tick := time.NewTicker(10 * time.Millisecond) - defer tick.Stop() - for { - wsService.Publish(event) - select { - case actual := <-received: - return actual - case <-tick.C: - case <-deadline: - t.Fatalf("event %s/%s never arrived", event.SourceID, event.Name) - } + actualEvent, ok := <-outputChannel + assert.Equal(t, ok, true) + assert.DeepEqual(t, model.Event{SourceID: actualEvent.SourceID, Name: actualEvent.Name, Properties: actualEvent.Properties}, expectedEvent) } }