diff --git a/service/action_service_websocket.go b/service/action_service_websocket.go index ea6848a..5bc9cd6 100644 --- a/service/action_service_websocket.go +++ b/service/action_service_websocket.go @@ -108,7 +108,7 @@ func (s *ActionServiceWS) Subscribe(sourceID string, names []string) (chan model s.subscriberChannels[sourceID] = make(map[string][]chan model.Action) } - c := make(chan model.Action, 1) + c := make(chan model.Action, subscriberBuffer) for _, name := range names { if s.subscriberChannels[sourceID][name] == nil { @@ -190,7 +190,7 @@ func (s *ActionServiceWS) Start(ctx *context.Context) { s.subscriberChannels = nil } - close(s.inboundChannel) + // not closed: a Publish/Trigger may hold it; senders select on ctx.Done, which ended the loop s.inboundChannel = nil close(s.stop) diff --git a/service/event_service_websocket.go b/service/event_service_websocket.go index e1cabd2..1c905c4 100644 --- a/service/event_service_websocket.go +++ b/service/event_service_websocket.go @@ -200,7 +200,7 @@ func (s *EventServiceWS) Start(ctx *context.Context) { s.subscriberChannels = nil } - close(s.inboundChannel) + // not closed: a Publish/Trigger may hold it; senders select on ctx.Done, which ended the loop s.inboundChannel = nil close(s.stop) diff --git a/service/event_type_service_test.go b/service/event_type_service_test.go index 95ba1e1..73bf609 100644 --- a/service/event_type_service_test.go +++ b/service/event_type_service_test.go @@ -3,6 +3,7 @@ package service import ( "context" "testing" + "time" "github.com/ReCasaOS/CasaOS-MessageBus/model" "github.com/ReCasaOS/CasaOS-MessageBus/repository" @@ -93,8 +94,14 @@ func TestEventTypeService(t *testing.T) { wsService.Publish(expectedEvent) - actualEvent, ok := <-outputChannel - assert.Equal(t, ok, true) + var actualEvent model.Event + select { + case received, ok := <-outputChannel: + assert.Equal(t, ok, true) + actualEvent = received + case <-time.After(10 * time.Second): + t.Fatal("the published event never arrived") + } assert.DeepEqual(t, model.Event{SourceID: actualEvent.SourceID, Name: actualEvent.Name, Properties: actualEvent.Properties}, expectedEvent) } } diff --git a/service/teardown_test.go b/service/teardown_test.go new file mode 100644 index 0000000..2d3f34f --- /dev/null +++ b/service/teardown_test.go @@ -0,0 +1,40 @@ +package service + +import ( + "context" + "sync" + "testing" + + "github.com/ReCasaOS/CasaOS-MessageBus/model" +) + +// Publish and Trigger racing the end of Start: the inbound channel is not closed under a sender +// (a send on a closed channel panics, and select may pick it while ctx.Done is ready too). +func TestSendersRacingTheEndOfStartDoNotPanic(t *testing.T) { + for range 300 { + events, actions := newEventService(t), newActionService(t) + ctx, cancel := context.WithCancel(context.Background()) + var stopped sync.WaitGroup + stopped.Add(2) + go func() { defer stopped.Done(); events.Start(&ctx) }() + go func() { defer stopped.Done(); actions.Start(&ctx) }() + <-events.Ready() + <-actions.Ready() + + var senders sync.WaitGroup + go cancel() // the senders and the end of Start at once + for range 64 { + senders.Add(2) + go func() { + defer senders.Done() + events.Publish(model.Event{SourceID: "app-management", Name: "backup:error"}) + }() + go func() { + defer senders.Done() + actions.Trigger(model.Action{SourceID: "app-management", Name: "app:install"}) + }() + } + senders.Wait() + stopped.Wait() + } +}