event/eventmanager.go

@@ -56,6 +56,7 @@ type manager struct {
 	subscribers map[Type][]Queue
 	events      chan []byte
 	mapMutex    sync.Mutex
+	chanMutex   sync.Mutex
 	internal    chan bool
 	closed      bool
 	trace       bool
@@ -97,6 +98,8 @@ func (em *manager) Subscribe(eventType Type, queue Queue) {
 
 // Publish takes an Event and sends it to the internal eventBus where it is distributed to all Subscribers
 func (em *manager) Publish(event Event) {
+	em.chanMutex.Lock()
+	defer em.chanMutex.Unlock()
 	if event.EventType != "" && !em.closed {
 
 		// Debug Events for Tracing, locked behind an environment variable
@@ -160,7 +163,9 @@ func (em *manager) eventBus() {
 // Shutdown triggers, and waits for, the internal eventBus goroutine to finish
 func (em *manager) Shutdown() {
 	em.events <- []byte{}
+	em.chanMutex.Lock()
 	em.closed = true
+	em.chanMutex.Unlock()
 	// wait for eventBus to finish
 	<-em.internal
 	close(em.events)