|
|
@@ -20,7 +20,13 @@ import (
|
|
|
"wails-app/internal/global"
|
|
|
)
|
|
|
|
|
|
-const cameraStreamTokenTTL = 2 * time.Minute
|
|
|
+const (
|
|
|
+ cameraStreamTokenTTL = 2 * time.Minute
|
|
|
+ // A stream has to emit data recently before it is considered online. The
|
|
|
+ // short grace window prevents an active MJPEG stream from briefly flipping
|
|
|
+ // to offline between browser refreshes.
|
|
|
+ cameraStreamOnlineTTL = 15 * time.Second
|
|
|
+)
|
|
|
|
|
|
type cameraStreamToken struct {
|
|
|
DeviceCode string
|
|
|
@@ -28,6 +34,214 @@ type cameraStreamToken struct {
|
|
|
}
|
|
|
|
|
|
var cameraStreamTokens sync.Map
|
|
|
+var cameraLastFrameAt sync.Map
|
|
|
+
|
|
|
+// cameraMJPEGHub keeps one FFmpeg source per camera. Browser pages can create
|
|
|
+// multiple MJPEG connections during refreshes, route transitions, or when the
|
|
|
+// workbench is open in more than one window; those connections must not each
|
|
|
+// open a separate RTSP session and transcoder.
|
|
|
+var cameraMJPEGHub = newCameraStreamHub()
|
|
|
+
|
|
|
+type cameraStreamConfig struct {
|
|
|
+ DeviceCode string
|
|
|
+ SourceURL string
|
|
|
+ FFmpegPath string
|
|
|
+ FPS int
|
|
|
+ Quality int
|
|
|
+}
|
|
|
+
|
|
|
+type cameraStreamSubscription struct {
|
|
|
+ frames <-chan []byte
|
|
|
+ unsubscribe func()
|
|
|
+}
|
|
|
+
|
|
|
+type cameraStreamHub struct {
|
|
|
+ mu sync.Mutex
|
|
|
+ sessions map[string]*cameraStreamSession
|
|
|
+}
|
|
|
+
|
|
|
+type cameraStreamSession struct {
|
|
|
+ hub *cameraStreamHub
|
|
|
+ config cameraStreamConfig
|
|
|
+ ctx context.Context
|
|
|
+ cancel context.CancelFunc
|
|
|
+
|
|
|
+ mu sync.Mutex
|
|
|
+ nextID uint64
|
|
|
+ subscribers map[uint64]chan []byte
|
|
|
+}
|
|
|
+
|
|
|
+func newCameraStreamHub() *cameraStreamHub {
|
|
|
+ return &cameraStreamHub{sessions: make(map[string]*cameraStreamSession)}
|
|
|
+}
|
|
|
+
|
|
|
+func newCameraStreamSession(hub *cameraStreamHub, config cameraStreamConfig) *cameraStreamSession {
|
|
|
+ ctx, cancel := context.WithCancel(context.Background())
|
|
|
+ return &cameraStreamSession{
|
|
|
+ hub: hub,
|
|
|
+ config: config,
|
|
|
+ ctx: ctx,
|
|
|
+ cancel: cancel,
|
|
|
+ subscribers: make(map[uint64]chan []byte),
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+func (h *cameraStreamHub) subscribe(config cameraStreamConfig) *cameraStreamSubscription {
|
|
|
+ h.mu.Lock()
|
|
|
+ session := h.sessions[config.DeviceCode]
|
|
|
+ created := session == nil
|
|
|
+ if session == nil {
|
|
|
+ session = newCameraStreamSession(h, config)
|
|
|
+ h.sessions[config.DeviceCode] = session
|
|
|
+ }
|
|
|
+ id, frames := session.addSubscriber()
|
|
|
+ if created {
|
|
|
+ go session.run()
|
|
|
+ }
|
|
|
+ h.mu.Unlock()
|
|
|
+ return &cameraStreamSubscription{
|
|
|
+ frames: frames,
|
|
|
+ unsubscribe: func() {
|
|
|
+ session.removeSubscriber(id)
|
|
|
+ },
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+func (h *cameraStreamHub) removeSession(session *cameraStreamSession) {
|
|
|
+ h.mu.Lock()
|
|
|
+ if h.sessions[session.config.DeviceCode] == session {
|
|
|
+ delete(h.sessions, session.config.DeviceCode)
|
|
|
+ }
|
|
|
+ h.mu.Unlock()
|
|
|
+}
|
|
|
+
|
|
|
+func (h *cameraStreamHub) stop(deviceCode string) {
|
|
|
+ h.mu.Lock()
|
|
|
+ session := h.sessions[deviceCode]
|
|
|
+ if session != nil {
|
|
|
+ delete(h.sessions, deviceCode)
|
|
|
+ }
|
|
|
+ h.mu.Unlock()
|
|
|
+ if session != nil {
|
|
|
+ session.cancel()
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+func (s *cameraStreamSession) addSubscriber() (uint64, <-chan []byte) {
|
|
|
+ s.mu.Lock()
|
|
|
+ defer s.mu.Unlock()
|
|
|
+ s.nextID++
|
|
|
+ frames := make(chan []byte, 64)
|
|
|
+ s.subscribers[s.nextID] = frames
|
|
|
+ return s.nextID, frames
|
|
|
+}
|
|
|
+
|
|
|
+func (s *cameraStreamSession) removeSubscriber(id uint64) {
|
|
|
+ s.mu.Lock()
|
|
|
+ delete(s.subscribers, id)
|
|
|
+ empty := len(s.subscribers) == 0
|
|
|
+ s.mu.Unlock()
|
|
|
+ if empty {
|
|
|
+ s.hub.removeSession(s)
|
|
|
+ s.cancel()
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+func (s *cameraStreamSession) publish(frame []byte) {
|
|
|
+ s.mu.Lock()
|
|
|
+ defer s.mu.Unlock()
|
|
|
+ for _, subscriber := range s.subscribers {
|
|
|
+ select {
|
|
|
+ case subscriber <- frame:
|
|
|
+ default:
|
|
|
+ // A slow client can resynchronize at the next MJPEG boundary. Never
|
|
|
+ // let it block the camera source or every other subscribed browser.
|
|
|
+ }
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+func (s *cameraStreamSession) closeSubscribers() {
|
|
|
+ s.mu.Lock()
|
|
|
+ for id, subscriber := range s.subscribers {
|
|
|
+ close(subscriber)
|
|
|
+ delete(s.subscribers, id)
|
|
|
+ }
|
|
|
+ s.mu.Unlock()
|
|
|
+}
|
|
|
+
|
|
|
+func (s *cameraStreamSession) run() {
|
|
|
+ defer s.closeSubscribers()
|
|
|
+ defer s.hub.removeSession(s)
|
|
|
+
|
|
|
+ args := []string{"-hide_banner", "-loglevel", "error"}
|
|
|
+ if strings.HasPrefix(strings.ToLower(s.config.SourceURL), "rtsp://") {
|
|
|
+ args = append(args, "-rtsp_transport", "tcp")
|
|
|
+ }
|
|
|
+ args = append(args,
|
|
|
+ "-i", s.config.SourceURL,
|
|
|
+ "-an",
|
|
|
+ "-vf", fmt.Sprintf("fps=%d", s.config.FPS),
|
|
|
+ "-q:v", fmt.Sprintf("%d", s.config.Quality),
|
|
|
+ "-f", "mpjpeg",
|
|
|
+ "pipe:1",
|
|
|
+ )
|
|
|
+
|
|
|
+ cmd := exec.CommandContext(s.ctx, s.config.FFmpegPath, args...)
|
|
|
+ if global.GVA_LOG != nil {
|
|
|
+ global.GVA_LOG.Info("启动共享摄像头 FFmpeg 转码", zap.String("device_code", s.config.DeviceCode), zap.String("ffmpeg", s.config.FFmpegPath), zap.String("source", redactCameraURL(s.config.SourceURL)))
|
|
|
+ }
|
|
|
+ stdout, err := cmd.StdoutPipe()
|
|
|
+ if err != nil {
|
|
|
+ if global.GVA_LOG != nil {
|
|
|
+ global.GVA_LOG.Error("创建摄像头 FFmpeg 输出失败", zap.String("device_code", s.config.DeviceCode), zap.Error(err))
|
|
|
+ }
|
|
|
+ return
|
|
|
+ }
|
|
|
+ cmd.Stderr = io.Discard
|
|
|
+ if err := cmd.Start(); err != nil {
|
|
|
+ if global.GVA_LOG != nil {
|
|
|
+ global.GVA_LOG.Error("摄像头 FFmpeg 启动失败", zap.String("device_code", s.config.DeviceCode), zap.String("ffmpeg", s.config.FFmpegPath), zap.Error(err))
|
|
|
+ }
|
|
|
+ return
|
|
|
+ }
|
|
|
+
|
|
|
+ buffer := make([]byte, 32*1024)
|
|
|
+ for {
|
|
|
+ n, readErr := stdout.Read(buffer)
|
|
|
+ if n > 0 {
|
|
|
+ frame := append([]byte(nil), buffer[:n]...)
|
|
|
+ markCameraStreamFrame(s.config.DeviceCode)
|
|
|
+ s.publish(frame)
|
|
|
+ }
|
|
|
+ if readErr != nil {
|
|
|
+ if !errors.Is(readErr, io.EOF) && s.ctx.Err() == nil && global.GVA_LOG != nil {
|
|
|
+ global.GVA_LOG.Warn("读取摄像头 FFmpeg 输出失败", zap.String("device_code", s.config.DeviceCode), zap.Error(readErr))
|
|
|
+ }
|
|
|
+ break
|
|
|
+ }
|
|
|
+ }
|
|
|
+ if err := cmd.Wait(); err != nil && s.ctx.Err() == nil && global.GVA_LOG != nil {
|
|
|
+ global.GVA_LOG.Error("摄像头 FFmpeg 转码进程退出", zap.String("device_code", s.config.DeviceCode), zap.Error(err))
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+func markCameraStreamFrame(deviceCode string) {
|
|
|
+ now := time.Now()
|
|
|
+ previous, hadPrevious := cameraLastFrameAt.Load(deviceCode)
|
|
|
+ cameraLastFrameAt.Store(deviceCode, now)
|
|
|
+ if (!hadPrevious || now.Sub(previous.(time.Time)) > cameraStreamOnlineTTL) && global.GVA_LOG != nil {
|
|
|
+ global.GVA_LOG.Info("摄像头视频流已在线", zap.String("device_code", deviceCode))
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+func hasRecentCameraStreamFrame(deviceCode string) bool {
|
|
|
+ value, ok := cameraLastFrameAt.Load(deviceCode)
|
|
|
+ if !ok {
|
|
|
+ return false
|
|
|
+ }
|
|
|
+ lastFrameAt, ok := value.(time.Time)
|
|
|
+ return ok && time.Since(lastFrameAt) <= cameraStreamOnlineTTL
|
|
|
+}
|
|
|
|
|
|
func isSystemMJPEG(protocol string) bool {
|
|
|
protocol = strings.ToLower(strings.TrimSpace(protocol))
|
|
|
@@ -45,6 +259,15 @@ func issueCameraStreamToken(deviceCode string) string {
|
|
|
DeviceCode: deviceCode,
|
|
|
ExpiresAt: time.Now().Add(cameraStreamTokenTTL),
|
|
|
})
|
|
|
+ // 每次页面刷新都会签发新令牌；过期项只在验证时才删除，长期运行会
|
|
|
+ // 无限累积，这里在签发时顺手清理。
|
|
|
+ now := time.Now()
|
|
|
+ cameraStreamTokens.Range(func(key, value any) bool {
|
|
|
+ if entry, ok := value.(cameraStreamToken); ok && now.After(entry.ExpiresAt) {
|
|
|
+ cameraStreamTokens.Delete(key)
|
|
|
+ }
|
|
|
+ return true
|
|
|
+ })
|
|
|
return token
|
|
|
}
|
|
|
}
|
|
|
@@ -63,7 +286,7 @@ func validateCameraStreamToken(deviceCode, token string) bool {
|
|
|
}
|
|
|
|
|
|
// StreamCameraMJPEG 在系统端把摄像头原始 RTSP/HTTP 流转换为 multipart MJPEG。
|
|
|
-// 该输出可由 WebView 的 <img> 直接播放，不要求边缘端实现 WHEP。
|
|
|
+// 同一摄像头的浏览器连接订阅同一个 FFmpeg 会话，避免重复转码耗尽主机资源。
|
|
|
func (s *PassageService) StreamCameraMJPEG(deviceCode, token string, writer http.ResponseWriter, ctx context.Context) error {
|
|
|
if !validateCameraStreamToken(deviceCode, token) {
|
|
|
return errors.New("摄像头播放令牌无效或已过期")
|
|
|
@@ -86,42 +309,15 @@ func (s *PassageService) StreamCameraMJPEG(deviceCode, token string, writer http
|
|
|
return errors.New("摄像头未配置原始视频地址")
|
|
|
}
|
|
|
|
|
|
- ffmpegPath := strings.TrimSpace(global.GVA_CONFIG.Camera.FFmpegPath)
|
|
|
- if ffmpegPath == "" {
|
|
|
- ffmpegPath = "ffmpeg"
|
|
|
- } else if info, statErr := os.Stat(ffmpegPath); statErr == nil && info.IsDir() {
|
|
|
- // Accept a configured FFmpeg bin directory for compatibility with older config files.
|
|
|
- ffmpegPath = filepath.Join(ffmpegPath, "ffmpeg.exe")
|
|
|
- }
|
|
|
- fps := global.GVA_CONFIG.Camera.MJPEGFPS
|
|
|
- if fps < 1 || fps > 25 {
|
|
|
- fps = 8
|
|
|
- }
|
|
|
- quality := global.GVA_CONFIG.Camera.JPEGQuality
|
|
|
- if quality < 2 || quality > 31 {
|
|
|
- quality = 5
|
|
|
- }
|
|
|
- args := []string{"-hide_banner", "-loglevel", "error"}
|
|
|
- if strings.HasPrefix(strings.ToLower(sourceURL), "rtsp://") {
|
|
|
- args = append(args, "-rtsp_transport", "tcp")
|
|
|
- }
|
|
|
- args = append(args, "-i", sourceURL, "-an", "-vf", fmt.Sprintf("fps=%d", fps), "-q:v", fmt.Sprintf("%d", quality), "-f", "mpjpeg", "pipe:1")
|
|
|
-
|
|
|
- cmd := exec.CommandContext(ctx, ffmpegPath, args...)
|
|
|
- if global.GVA_LOG != nil {
|
|
|
- global.GVA_LOG.Info("启动摄像头 FFmpeg 转码", zap.String("device_code", deviceCode), zap.String("ffmpeg", ffmpegPath), zap.String("source", redactCameraURL(sourceURL)))
|
|
|
- }
|
|
|
- stdout, err := cmd.StdoutPipe()
|
|
|
- if err != nil {
|
|
|
- return fmt.Errorf("创建视频转换输出失败: %w", err)
|
|
|
- }
|
|
|
- cmd.Stderr = io.Discard
|
|
|
- if err := cmd.Start(); err != nil {
|
|
|
- if global.GVA_LOG != nil {
|
|
|
- global.GVA_LOG.Error("摄像头 FFmpeg 启动失败", zap.String("device_code", deviceCode), zap.String("ffmpeg", ffmpegPath), zap.Error(err))
|
|
|
- }
|
|
|
- return fmt.Errorf("启动 FFmpeg 失败（%s）: %w", ffmpegPath, err)
|
|
|
+ config := cameraStreamConfig{
|
|
|
+ DeviceCode: deviceCode,
|
|
|
+ SourceURL: sourceURL,
|
|
|
+ FFmpegPath: resolveFFmpegPath(),
|
|
|
+ FPS: configuredMJPEGFPS(),
|
|
|
+ Quality: configuredJPEGQuality(),
|
|
|
}
|
|
|
+ subscription := cameraMJPEGHub.subscribe(config)
|
|
|
+ defer subscription.unsubscribe()
|
|
|
|
|
|
writer.Header().Set("Content-Type", "multipart/x-mixed-replace; boundary=ffmpeg")
|
|
|
writer.Header().Set("Cache-Control", "no-store, no-cache, must-revalidate")
|
|
|
@@ -129,39 +325,55 @@ func (s *PassageService) StreamCameraMJPEG(deviceCode, token string, writer http
|
|
|
if flusher, ok := writer.(http.Flusher); ok {
|
|
|
flusher.Flush()
|
|
|
}
|
|
|
- _, copyErr := io.Copy(&flushWriter{writer: writer}, stdout)
|
|
|
- waitErr := cmd.Wait()
|
|
|
- if copyErr != nil && !errors.Is(copyErr, context.Canceled) {
|
|
|
- return fmt.Errorf("视频流传输失败: %w", copyErr)
|
|
|
+ for {
|
|
|
+ select {
|
|
|
+ case <-ctx.Done():
|
|
|
+ return nil
|
|
|
+ case frame, ok := <-subscription.frames:
|
|
|
+ if !ok {
|
|
|
+ return nil
|
|
|
+ }
|
|
|
+ if _, err := writer.Write(frame); err != nil {
|
|
|
+ return nil
|
|
|
+ }
|
|
|
+ if flusher, ok := writer.(http.Flusher); ok {
|
|
|
+ flusher.Flush()
|
|
|
+ }
|
|
|
+ }
|
|
|
}
|
|
|
- if ctx.Err() != nil {
|
|
|
- return nil
|
|
|
+}
|
|
|
+
|
|
|
+func resolveFFmpegPath() string {
|
|
|
+ ffmpegPath := strings.TrimSpace(global.GVA_CONFIG.Camera.FFmpegPath)
|
|
|
+ if ffmpegPath == "" {
|
|
|
+ return "ffmpeg"
|
|
|
}
|
|
|
- if waitErr != nil {
|
|
|
- if global.GVA_LOG != nil {
|
|
|
- global.GVA_LOG.Error("摄像头 FFmpeg 转码进程退出", zap.String("device_code", deviceCode), zap.Error(waitErr))
|
|
|
- }
|
|
|
- return fmt.Errorf("FFmpeg 进程退出: %w", waitErr)
|
|
|
+ if info, err := os.Stat(ffmpegPath); err == nil && info.IsDir() {
|
|
|
+ return filepath.Join(ffmpegPath, "ffmpeg.exe")
|
|
|
}
|
|
|
- return nil
|
|
|
+ return ffmpegPath
|
|
|
}
|
|
|
|
|
|
-func redactCameraURL(raw string) string {
|
|
|
- u, err := url.Parse(raw)
|
|
|
- if err != nil || u.User == nil {
|
|
|
- return raw
|
|
|
+func configuredMJPEGFPS() int {
|
|
|
+ fps := global.GVA_CONFIG.Camera.MJPEGFPS
|
|
|
+ if fps < 1 || fps > 25 {
|
|
|
+ return 8
|
|
|
}
|
|
|
- return strings.Replace(raw, u.User.String()+"@", "***@", 1)
|
|
|
+ return fps
|
|
|
}
|
|
|
|
|
|
-type flushWriter struct {
|
|
|
- writer http.ResponseWriter
|
|
|
+func configuredJPEGQuality() int {
|
|
|
+ quality := global.GVA_CONFIG.Camera.JPEGQuality
|
|
|
+ if quality < 2 || quality > 31 {
|
|
|
+ return 5
|
|
|
+ }
|
|
|
+ return quality
|
|
|
}
|
|
|
|
|
|
-func (w *flushWriter) Write(data []byte) (int, error) {
|
|
|
- n, err := w.writer.Write(data)
|
|
|
- if flusher, ok := w.writer.(http.Flusher); ok {
|
|
|
- flusher.Flush()
|
|
|
+func redactCameraURL(raw string) string {
|
|
|
+ u, err := url.Parse(raw)
|
|
|
+ if err != nil || u.User == nil {
|
|
|
+ return raw
|
|
|
}
|
|
|
- return n, err
|
|
|
+ return strings.Replace(raw, u.User.String()+"@", "***@", 1)
|
|
|
}
|
