监控网页实现

backend/internal/stream/server.go

@@ -17,6 +17,7 @@ import (
 	"hightube/internal/chat"
 	"hightube/internal/db"
 	"hightube/internal/model"
+	"hightube/internal/monitor"
 )
 
 func init() {
@@ -51,12 +52,12 @@ func NewRTMPServer() *RTMPServer {
 	// Triggered when a broadcaster (e.g., OBS) starts publishing
 	s.server.HandlePublish = func(conn *rtmp.Conn) {
 		streamPath := conn.URL.Path // Expected format: /live/{stream_key}
-		fmt.Printf("[INFO] OBS is attempting to publish to: %s\n", streamPath)
+		monitor.Infof("OBS publish attempt: %s", streamPath)
 
 		// Extract stream key from path
 		parts := strings.Split(streamPath, "/")
 		if len(parts) < 3 || parts[1] != "live" {
-			fmt.Printf("[WARN] Invalid publish path format: %s\n", streamPath)
+			monitor.Warnf("Invalid publish path format: %s", streamPath)
 			return
 		}
 		streamKey := parts[2]
@@ -64,16 +65,16 @@ func NewRTMPServer() *RTMPServer {
 		// Authenticate stream key
 		var room model.Room
 		if err := db.DB.Where("stream_key = ?", streamKey).First(&room).Error; err != nil {
-			fmt.Printf("[WARN] Authentication failed, invalid stream key: %s\n", streamKey)
+			monitor.Warnf("Invalid stream key: %s", streamKey)
 			return // Reject connection
 		}
 
-		fmt.Printf("[INFO] Stream authenticated for Room ID: %d\n", room.ID)
+		monitor.Infof("Stream authenticated for room_id=%d", room.ID)
 
 		// 1. Get audio/video stream metadata
 		streams, err := conn.Streams()
 		if err != nil {
-			fmt.Printf("[ERROR] Failed to parse stream headers: %v\n", err)
+			monitor.Errorf("Failed to parse stream headers: %v", err)
 			return
 		}
 
@@ -101,7 +102,7 @@ func NewRTMPServer() *RTMPServer {
 			// Clear chat history for this room
 			chat.MainHub.ClearRoomHistory(fmt.Sprintf("%d", room.ID))
 			
-			fmt.Printf("[INFO] Publishing ended for Room ID: %d\n", room.ID)
+			monitor.Infof("Publishing ended for room_id=%d", room.ID)
 		}()
 
 		// 4. Continuously copy data packets to our broadcast queue
@@ -111,7 +112,7 @@ func NewRTMPServer() *RTMPServer {
 	// Triggered when a viewer (e.g., VLC) requests playback
 	s.server.HandlePlay = func(conn *rtmp.Conn) {
 		streamPath := conn.URL.Path // Expected format: /live/{room_id}
-		fmt.Printf("[INFO] VLC is pulling stream from: %s\n", streamPath)
+		monitor.Infof("RTMP play requested: %s", streamPath)
 
 		// 1. Look for the requested room's data queue
 		s.mutex.RLock()
@@ -119,7 +120,7 @@ func NewRTMPServer() *RTMPServer {
 		s.mutex.RUnlock()
 
 		if !ok {
-			fmt.Printf("[WARN] Stream not found or inactive: %s\n", streamPath)
+			monitor.Warnf("Stream not found or inactive: %s", streamPath)
 			return
 		}
 
@@ -129,7 +130,7 @@ func NewRTMPServer() *RTMPServer {
 		conn.WriteHeader(streams)
 
 		// 3. Cleanup on end
-		defer fmt.Printf("[INFO] Playback ended: %s\n", streamPath)
+		defer monitor.Infof("Playback ended: %s", streamPath)
 
 		// 4. Continuously copy data packets to the viewer
 		err := avutil.CopyPackets(conn, cursor)
@@ -137,9 +138,9 @@ func NewRTMPServer() *RTMPServer {
 			// 如果是客户端主动断开连接引起的错误，不将其作为严重错误打印
 			errStr := err.Error()
 			if strings.Contains(errStr, "broken pipe") || strings.Contains(errStr, "connection reset by peer") {
-				fmt.Printf("[INFO] Viewer disconnected normally: %s\n", streamPath)
+				monitor.Infof("Viewer disconnected: %s", streamPath)
 			} else {
-				fmt.Printf("[ERROR] Error occurred during playback: %v\n", err)
+				monitor.Errorf("Playback error on %s: %v", streamPath, err)
 			}
 		}
 	}
@@ -150,7 +151,7 @@ func NewRTMPServer() *RTMPServer {
 // Start launches the RTMP server
 func (s *RTMPServer) Start(addr string) error {
 	s.server.Addr = addr
-	fmt.Printf("[INFO] RTMP Server is listening on %s...\n", addr)
+	monitor.Infof("RTMP server listening on %s", addr)
 	return s.server.ListenAndServe()
 }
 
@@ -190,9 +191,26 @@ func (s *RTMPServer) HandleHTTPFLV(c *gin.Context) {
 	if err := avutil.CopyFile(muxer, cursor); err != nil && err != io.EOF {
 		errStr := err.Error()
 		if strings.Contains(errStr, "broken pipe") || strings.Contains(errStr, "connection reset by peer") {
-			fmt.Printf("[INFO] HTTP-FLV viewer disconnected normally: %s\n", streamPath)
+			monitor.Infof("HTTP-FLV viewer disconnected: %s", streamPath)
 			return
 		}
-		fmt.Printf("[ERROR] HTTP-FLV playback error on %s: %v\n", streamPath, err)
+		monitor.Errorf("HTTP-FLV playback error on %s: %v", streamPath, err)
 	}
 }
+
+func (s *RTMPServer) ActiveStreamCount() int {
+	s.mutex.RLock()
+	defer s.mutex.RUnlock()
+	return len(s.channels)
+}
+
+func (s *RTMPServer) ActiveStreamPaths() []string {
+	s.mutex.RLock()
+	defer s.mutex.RUnlock()
+
+	paths := make([]string, 0, len(s.channels))
+	for path := range s.channels {
+		paths = append(paths, path)
+	}
+	return paths
+}