fix(service): Start's teardown does not close the inbound channel (follow-up of #4)

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)

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)

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)
 	}
 }

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()
+	}
+}