Update CHANGELOG.md

Only attempt to close channels once, rather than per topic
they are subscribed to.

CHANGELOG.md

@@ -1,5 +1,11 @@
 # Changelog
 
+## v1.5.5
+### Fixed
+* Fixes sending to a closed channel when sending server logs over the websocket
+* Fixes `wings diagnostics` uploading no content
+* Fixes a panic caused by the event bus closing channels multiple times when a server is deleted
+
 ## v1.5.4
 ### Fixed
 * Fixes SSL paths being improperly converted to lowercase in environments where the path is case-sensitive.

cmd/diagnostics.go

@@ -179,6 +179,17 @@ func diagnosticsCmdRun(cmd *cobra.Command, args []string) {
 		fmt.Fprintln(output, "Logs redacted.")
 	}
 
+	if !diagnosticsArgs.IncludeEndpoints {
+		s := output.String()
+		output.Reset()
+		s = strings.ReplaceAll(s, cfg.PanelLocation, "{redacted}")
+		s = strings.ReplaceAll(s, cfg.Api.Host, "{redacted}")
+		s = strings.ReplaceAll(s, cfg.Api.Ssl.CertificateFile, "{redacted}")
+		s = strings.ReplaceAll(s, cfg.Api.Ssl.KeyFile, "{redacted}")
+		s = strings.ReplaceAll(s, cfg.System.Sftp.Address, "{redacted}")
+		output.WriteString(s)
+	}
+
 	fmt.Println("\n---------------  generated report  ---------------")
 	fmt.Println(output.String())
 	fmt.Print("---------------   end of report    ---------------\n\n")
@@ -188,16 +199,6 @@ func diagnosticsCmdRun(cmd *cobra.Command, args []string) {
 		survey.AskOne(&survey.Confirm{Message: "Upload to " + diagnosticsArgs.HastebinURL + "?", Default: false}, &upload)
 	}
 	if upload {
-		if !diagnosticsArgs.IncludeEndpoints {
-			s := output.String()
-			output.Reset()
-			a := strings.ReplaceAll(cfg.PanelLocation, s, "{redacted}")
-			a = strings.ReplaceAll(cfg.Api.Host, a, "{redacted}")
-			a = strings.ReplaceAll(cfg.Api.Ssl.CertificateFile, a, "{redacted}")
-			a = strings.ReplaceAll(cfg.Api.Ssl.KeyFile, a, "{redacted}")
-			a = strings.ReplaceAll(cfg.System.Sftp.Address, a, "{redacted}")
-			output.WriteString(a)
-		}
 		u, err := uploadToHastebin(diagnosticsArgs.HastebinURL, output.String())
 		if err == nil {
 			fmt.Println("Your report is available here: ", u)

events/events.go

@@ -31,8 +31,14 @@ func (b *Bus) Off(listener Listener, topics ...string) {
 	b.listenersMx.Lock()
 	defer b.listenersMx.Unlock()
 
+	var closed bool
+
 	for _, topic := range topics {
-		b.off(topic, listener)
+		ok := b.off(topic, listener)
+		if !closed && ok {
+			close(listener)
+			closed = true
+		}
 	}
 }
 
@@ -116,11 +122,30 @@ func (b *Bus) Destroy() {
 	b.listenersMx.Lock()
 	defer b.listenersMx.Unlock()
 
+	// Track what listeners have already been closed. Because the same listener
+	// can be listening on multiple topics, we need a way to essentially
+	// "de-duplicate" all the listeners across all the topics.
+	var closed []Listener
+
 	for _, listeners := range b.listeners {
 		for _, listener := range listeners {
+			if contains(closed, listener) {
+				continue
+			}
+
 			close(listener)
+			closed = append(closed, listener)
 		}
 	}
 
 	b.listeners = make(map[string][]Listener)
 }
+
+func contains(closed []Listener, listener Listener) bool {
+	for _, c := range closed {
+		if c == listener {
+			return true
+		}
+	}
+	return false
+}

events/events_test.go

@@ -36,8 +36,6 @@ func TestBus_Off(t *testing.T) {
 
 			bus.Off(listener, topic)
 			g.Assert(len(bus.listeners[topic])).Equal(0, "Topic still has one or more listeners")
-
-			close(listener)
 		})
 
 		g.It("unregisters correct listener", func() {
@@ -62,9 +60,6 @@ func TestBus_Off(t *testing.T) {
 
 			// Cleanup
 			bus.Off(listener2, topic)
-			close(listener)
-			close(listener2)
-			close(listener3)
 		})
 	})
 }
@@ -91,7 +86,6 @@ func TestBus_On(t *testing.T) {
 
 			// Cleanup
 			bus.Off(listener, topic)
-			close(listener)
 		})
 	})
 }
@@ -127,7 +121,6 @@ func TestBus_Publish(t *testing.T) {
 			<-done
 
 			// Cleanup
-			close(listener)
 			bus.Off(listener, topic)
 		})
 
@@ -172,9 +165,6 @@ func TestBus_Publish(t *testing.T) {
 			bus.Off(listener, topic)
 			bus.Off(listener2, topic)
 			bus.Off(listener3, topic)
-			close(listener)
-			close(listener2)
-			close(listener3)
 		})
 	})
 }

router/websocket/listeners.go

@@ -146,12 +146,10 @@ func (h *Handler) listenForServerEvents(ctx context.Context) error {
 		break
 	}
 
+	// These functions will automatically close the channel if it hasn't been already.
 	h.server.Events().Off(eventChan, e...)
-	h.server.InstallSink().Off(logOutput)
+	h.server.LogSink().Off(logOutput)
 	h.server.InstallSink().Off(installOutput)
-	close(eventChan)
-	close(logOutput)
-	close(installOutput)
 
 	// If the internal context is stopped it is either because the parent context
 	// got canceled or because we ran into an error. If the "err" variable is nil

server/sink.go

@@ -16,14 +16,6 @@ func newSinkPool() *sinkPool {
 	return &sinkPool{}
 }
 
-// On adds a sink on the pool.
-func (p *sinkPool) On(c chan []byte) {
-	p.mx.Lock()
-	defer p.mx.Unlock()
-
-	p.sinks = append(p.sinks, c)
-}
-
 // Off removes a sink from the pool.
 func (p *sinkPool) Off(c chan []byte) {
 	p.mx.Lock()
@@ -39,10 +31,19 @@ func (p *sinkPool) Off(c chan []byte) {
 		sinks[len(sinks)-1] = nil
 		sinks = sinks[:len(sinks)-1]
 		p.sinks = sinks
+		close(c)
 		return
 	}
 }
 
+// On adds a sink on the pool.
+func (p *sinkPool) On(c chan []byte) {
+	p.mx.Lock()
+	defer p.mx.Unlock()
+
+	p.sinks = append(p.sinks, c)
+}
+
 // Destroy destroys the pool by removing and closing all sinks.
 func (p *sinkPool) Destroy() {
 	p.mx.Lock()