websocket: remove channel buffers

events/events.go

@@ -104,17 +104,13 @@ func (b *Bus) Publish(topic string, data interface{}) {
 		return
 	}
 
-	var wg sync.WaitGroup
 	event := Event{Topic: topic, Data: data}
 	for _, listener := range listeners {
 		l := listener
-		wg.Add(1)
 		go func(l Listener, event Event) {
-			defer wg.Done()
 			l <- event
 		}(l, event)
 	}
-	wg.Wait()
 }
 
 // Destroy destroys the Event Bus by unregistering and closing all listeners.

events/events_test.go

@@ -168,3 +168,58 @@ func TestBus_Publish(t *testing.T) {
 		})
 	})
 }
+
+func TestBus_Destroy(t *testing.T) {
+	g := Goblin(t)
+
+	g.Describe("Destroy", func() {
+		g.It("unsubscribes and closes all listeners", func() {
+			bus := NewBus()
+
+			listener := make(chan Event)
+			bus.On(listener, "test")
+
+			done := make(chan struct{}, 1)
+			go func() {
+				select {
+				case m := <-listener:
+					g.Assert(m).IsZero()
+				case <-time.After(1 * time.Second):
+					g.Fail("listener did not receive message in time")
+				}
+				done <- struct{}{}
+			}()
+			bus.Destroy()
+			<-done
+
+			g.Assert(bus.listeners).Equal(map[string][]Listener{})
+		})
+
+		// This is a check that ensures Destroy only closes each listener
+		// channel once, even if it is subscribed to multiple topics.
+		//
+		// Closing a channel multiple times will cause a runtime panic, which
+		// I'm pretty sure we don't want.
+		g.It("unsubscribes and closes channel only once", func() {
+			bus := NewBus()
+
+			listener := make(chan Event)
+			bus.On(listener, "test", "test2", "test3", "test4", "test5")
+
+			done := make(chan struct{}, 1)
+			go func() {
+				select {
+				case m := <-listener:
+					g.Assert(m).IsZero()
+				case <-time.After(1 * time.Second):
+					g.Fail("listener did not receive message in time")
+				}
+				done <- struct{}{}
+			}()
+			bus.Destroy()
+			<-done
+
+			g.Assert(bus.listeners).Equal(map[string][]Listener{})
+		})
+	})
+}

router/websocket/websocket.go

@@ -267,11 +267,18 @@ func (h *Handler) setJwt(token *tokens.WebsocketPayload) {
 	h.Unlock()
 }
 
+var actions = map[server.PowerAction]string{
+	server.PowerActionStart:     PermissionSendPowerStart,
+	server.PowerActionStop:      PermissionSendPowerStop,
+	server.PowerActionRestart:   PermissionSendPowerRestart,
+	server.PowerActionTerminate: PermissionSendPowerStop,
+}
+
 // HandleInbound handles an inbound socket request and route it to the proper action.
 func (h *Handler) HandleInbound(ctx context.Context, m Message) error {
 	if m.Event != AuthenticationEvent {
 		if err := h.TokenValid(); err != nil {
-			h.unsafeSendJson(Message{
+			_ = h.unsafeSendJson(Message{
 				Event: JwtErrorEvent,
 				Args:  []string{err.Error()},
 			})
@@ -339,12 +346,6 @@ func (h *Handler) HandleInbound(ctx context.Context, m Message) error {
 		{
 			action := server.PowerAction(strings.Join(m.Args, ""))
 
-			actions := make(map[server.PowerAction]string)
-			actions[server.PowerActionStart] = PermissionSendPowerStart
-			actions[server.PowerActionStop] = PermissionSendPowerStop
-			actions[server.PowerActionRestart] = PermissionSendPowerRestart
-			actions[server.PowerActionTerminate] = PermissionSendPowerStop
-
 			// Check that they have permission to perform this action if it is needed.
 			if permission, exists := actions[action]; exists {
 				if !h.GetJwt().HasPermission(permission) {

server/sink.go

@@ -2,7 +2,6 @@ package server
 
 import (
 	"sync"
-	"time"
 )
 
 // sinkPool represents a pool with sinks.
@@ -60,12 +59,9 @@ func (p *sinkPool) Destroy() {
 func (p *sinkPool) Push(v []byte) {
 	p.mx.RLock()
 	for _, c := range p.sinks {
-		// TODO: should this be done in parallel?
 		select {
 		// Send the log output to the channel
 		case c <- v:
-		// Timeout after 100 milliseconds, this will cause the write to the channel to be cancelled.
-		case <-time.After(100 * time.Millisecond):
 		}
 	}
 	p.mx.RUnlock()