network enhanecment / refactor (#361)

* chore(network): improve connectivity check * refactor(network): rewrite network and timesync component * feat(display): show cloud connection status * chore: change logging verbosity * chore(websecure): update log message * fix(ota): validate root certificate when downloading update * feat(ui): add network settings tab * fix(display): cloud connecting animation * fix: golintci issues * feat: add network settings tab * feat(timesync): query servers in parallel * refactor(network): move to internal/network package * feat(timesync): add metrics * refactor(log): move log to internal/logging package * refactor(mdms): move mdns to internal/mdns package * feat(developer): add pprof endpoint * feat(logging): add a simple logging streaming endpoint * fix(mdns): do not start mdns until network is up * feat(network): allow users to update network settings from ui * fix(network): handle errors when net.IPAddr is nil * fix(mdns): scopedLogger SIGSEGV * fix(dhcp): watch directory instead of file to catch fsnotify.Create event * refactor(nbd): move platform-specific code to different files * refactor(native): move platform-specific code to different files * chore: fix linter issues * chore(dev_deploy): allow to override PION_LOG_TRACE
2026-01-18 11:38:32 +01:00 · 2025-04-16 01:39:23 +02:00
parent 2b2a14204d
commit 189b84380b
71 changed files with 4938 additions and 825 deletions
--- a/internal/logging/sse.go
+++ b/internal/logging/sse.go
@@ -0,0 +1,137 @@
+package logging
+
+import (
+	"embed"
+	"io"
+	"net/http"
+
+	"github.com/gin-gonic/gin"
+	"github.com/rs/zerolog"
+)
+
+//go:embed sse.html
+var sseHTML embed.FS
+
+type sseEvent struct {
+	Message       chan string
+	NewClients    chan chan string
+	ClosedClients chan chan string
+	TotalClients  map[chan string]bool
+}
+
+// New event messages are broadcast to all registered client connection channels
+type sseClientChan chan string
+
+var (
+	sseServer *sseEvent
+	sseLogger *zerolog.Logger
+)
+
+func init() {
+	sseServer = newSseServer()
+	sseLogger = GetSubsystemLogger("sse")
+}
+
+// Initialize event and Start procnteessing requests
+func newSseServer() (event *sseEvent) {
+	event = &sseEvent{
+		Message:       make(chan string),
+		NewClients:    make(chan chan string),
+		ClosedClients: make(chan chan string),
+		TotalClients:  make(map[chan string]bool),
+	}
+
+	go event.listen()
+
+	return
+}
+
+// It Listens all incoming requests from clients.
+// Handles addition and removal of clients and broadcast messages to clients.
+func (stream *sseEvent) listen() {
+	for {
+		select {
+		// Add new available client
+		case client := <-stream.NewClients:
+			stream.TotalClients[client] = true
+			sseLogger.Info().
+				Int("total_clients", len(stream.TotalClients)).
+				Msg("new client connected")
+
+		// Remove closed client
+		case client := <-stream.ClosedClients:
+			delete(stream.TotalClients, client)
+			close(client)
+			sseLogger.Info().Int("total_clients", len(stream.TotalClients)).Msg("client disconnected")
+
+		// Broadcast message to client
+		case eventMsg := <-stream.Message:
+			for clientMessageChan := range stream.TotalClients {
+				select {
+				case clientMessageChan <- eventMsg:
+					// Message sent successfully
+				default:
+					// Failed to send, dropping message
+				}
+			}
+		}
+	}
+}
+
+func (stream *sseEvent) serveHTTP() gin.HandlerFunc {
+	return func(c *gin.Context) {
+		clientChan := make(sseClientChan)
+		stream.NewClients <- clientChan
+
+		go func() {
+			<-c.Writer.CloseNotify()
+
+			for range clientChan {
+			}
+
+			stream.ClosedClients <- clientChan
+		}()
+
+		c.Set("clientChan", clientChan)
+		c.Next()
+	}
+}
+
+func sseHeadersMiddleware() gin.HandlerFunc {
+	return func(c *gin.Context) {
+		if c.Request.Method == "GET" && c.NegotiateFormat(gin.MIMEHTML) == gin.MIMEHTML {
+			c.FileFromFS("/sse.html", http.FS(sseHTML))
+			c.Status(http.StatusOK)
+			c.Abort()
+			return
+		}
+
+		c.Writer.Header().Set("Content-Type", "text/event-stream")
+		c.Writer.Header().Set("Cache-Control", "no-cache")
+		c.Writer.Header().Set("Connection", "keep-alive")
+		c.Writer.Header().Set("Transfer-Encoding", "chunked")
+		c.Writer.Header().Set("Access-Control-Allow-Origin", "*")
+		c.Next()
+	}
+}
+
+func AttachSSEHandler(router *gin.RouterGroup) {
+	router.StaticFS("/log-stream", http.FS(sseHTML))
+	router.GET("/log-stream", sseHeadersMiddleware(), sseServer.serveHTTP(), func(c *gin.Context) {
+		v, ok := c.Get("clientChan")
+		if !ok {
+			return
+		}
+		clientChan, ok := v.(sseClientChan)
+		if !ok {
+			return
+		}
+		c.Stream(func(w io.Writer) bool {
+			if msg, ok := <-clientChan; ok {
+				c.SSEvent("message", msg)
+				return true
+			}
+			return false
+		})
+	})
+}