feat: multi-probe UX improvements

main.go

@@ -12,6 +12,7 @@ import (
 	"strconv"
 	"sync"
 	"syscall"
+	"time"
 
 	"github.com/gin-gonic/gin"
 	"github.com/google/uuid"
@@ -30,6 +31,8 @@ var (
 		WriteBufferSize: 1024,
 		CheckOrigin:     func(r *http.Request) bool { return true },
 	}
+
+	ProbeStats []*ProbeChannelStatus
 )
 
 type WebsocketConnection struct {
@@ -53,6 +56,12 @@ type DestinationUpdate struct {
 	Source SourceUpdate `json:"source"`
 }
 
+type ProbeChannelStatus struct {
+	Id           int  `json:"id"`
+	ActiveSource bool `json:"active_source"`
+	Clients      int  `json:"clients"`
+}
+
 type SourceUpdate struct {
 	Id    int    `json:"id"`
 	Label string `json:"label"`
@@ -92,8 +101,16 @@ func main() {
 
 	if Config.Probe.Enabled {
 		ProbeHandlers = make([]*ProbeSocketHandler, len(Config.Probe.RouterDestinations))
+		ProbeStats = make([]*ProbeChannelStatus, len(Config.Probe.RouterDestinations))
+		for i := range ProbeStats {
+			ProbeStats[i] = &ProbeChannelStatus{
+				Id: i,
+			}
+		}
+
 		for i := range Config.Probe.RouterDestinations {
 			ProbeHandlers[i] = &ProbeSocketHandler{
+				Id:         i,
 				clients:    make(map[*ProbeClient]bool),
 				register:   make(chan *ProbeClient),
 				unregister: make(chan *ProbeClient),
@@ -129,6 +146,20 @@ func main() {
 	log.Println("Exiting")
 }
 
+func SendProbeStats() {
+	payload, _ := json.Marshal(ProbeStats)
+	update := MatrixWSMessage{
+		Type: "probe_stats",
+		Data: payload,
+	}
+
+	MatrixWSConnectionsMutex.Lock()
+	for conn := range MatrixWSConnections {
+		conn.Socket.WriteJSON(update)
+	}
+	MatrixWSConnectionsMutex.Unlock()
+}
+
 func serveHTTP() {
 	gin.SetMode(gin.ReleaseMode)
 
@@ -255,6 +286,12 @@ func HandleMatrixWS(c *gin.Context) {
 	MatrixWSConnectionsMutex.Unlock()
 
 	defer connection.Socket.Close()
+
+	go func() {
+		time.Sleep(500 * time.Millisecond)
+		SendProbeStats()
+	}()
+
 	for {
 		_, messageBytes, err := connection.Socket.ReadMessage()
 		if err != nil {
@@ -300,6 +337,9 @@ func HandleProbeStream(c *gin.Context) {
 
 	log.Printf("stream for probe %d connected from %s", index, c.RemoteIP())
 
+	ProbeStats[index].ActiveSource = true
+	SendProbeStats()
+
 	for {
 		data, err := ioutil.ReadAll(io.LimitReader(c.Request.Body, 1024))
 		if err != nil || len(data) == 0 {
@@ -309,5 +349,8 @@ func HandleProbeStream(c *gin.Context) {
 		ProbeHandlers[index].BroadcastData(&data)
 	}
 
+	ProbeStats[index].ActiveSource = false
+	SendProbeStats()
+
 	log.Printf("stream for probe %d disconnected from %s", index, c.RemoteIP())
 }

probe.go

@@ -87,6 +87,7 @@ func (c *ProbeClient) Run() {
 }
 
 type ProbeSocketHandler struct {
+	Id         int
 	clients    map[*ProbeClient]bool // *client -> is connected (true/false)
 	register   chan *ProbeClient
 	unregister chan *ProbeClient
@@ -102,14 +103,18 @@ func (h *ProbeSocketHandler) Run() {
 		select {
 		case client := <-h.register:
 			h.clients[client] = true
-			log.Printf("New client registered. Total: %d\n", len(h.clients))
+			log.Printf("[probe %d] client connected. active clients: %d\n", h.Id, len(h.clients))
+			ProbeStats[h.Id].Clients = len(h.clients)
+			SendProbeStats()
 
 		case client := <-h.unregister:
 			_, ok := h.clients[client]
 			if ok {
 				delete(h.clients, client)
 			}
-			log.Printf("Client unregistered.   Total: %d\n", len(h.clients))
+			log.Printf("[probe %d] client disconnected. active clients: %d\n", h.Id, len(h.clients))
+			ProbeStats[h.Id].Clients = len(h.clients)
+			SendProbeStats()
 
 		case data := <-h.broadcast:
 			h.BroadcastData(data)
@@ -132,7 +137,6 @@ func (h *ProbeSocketHandler) ServeWS(c *gin.Context) {
 		return
 	}
 
-	log.Println("New client connected")
 	client := NewProbeClient(ws, h.unregister)
 
 	h.register <- client

ui/.eslintrc.js

@@ -26,4 +26,4 @@ module.exports = {
       { allowAsProps: true },
     ],
   },
-};
+};