ituoke/internal/device/stream.go

package device

import (
	"bufio"
	"bytes"
	"encoding/binary"
	"fmt"
	"io"
	"net"
	"net/http"
	"os"
	"os/exec"
	"path/filepath"
	"strings"
	"sync"
	"sync/atomic"
	"time"
)

// ============================================================
// [投屏流 · 关键] scrcpy-server 直连（H.264）→ ffmpeg → MJPEG
// 预览地址：http://127.0.0.1:<port>/mjpeg
// ============================================================

type previewStream struct {
	lifecycle sync.Mutex
	workers   sync.WaitGroup
	mu        sync.Mutex
	running   bool
	serial    string
	url       string

	serverCmd   *exec.Cmd
	ffmpegCmd   *exec.Cmd
	httpSrv     *http.Server
	httpLn      net.Listener
	controlConn net.Conn
	touchDown   bool
	lastTouch   TouchRequest
	videoConn   net.Conn
	videoLn     net.Listener

	latest  atomic.Value // []byte jpeg
	clients sync.Map
	stopCh  chan struct{}
}

func (s *Service) ensureStream() *previewStream {
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.stream == nil {
		s.stream = &previewStream{}
	}
	return s.stream
}

// StartPreviewStream 启动低延迟预览流
func (s *Service) StartPreviewStream(deviceID string) ActionResult {
	if deviceID == "" {
		s.mu.RLock()
		deviceID = s.active
		s.mu.RUnlock()
	}
	if deviceID == "" {
		return ActionResult{OK: false, Message: "没有选中的设备"}
	}

	st := s.ensureStream()
	st.lifecycle.Lock()
	defer st.lifecycle.Unlock()
	st.mu.Lock()
	if st.running {
		if st.serial == deviceID {
			u := st.url
			st.mu.Unlock()
			return ActionResult{OK: true, Message: u}
		}
		st.stopLocked()
	}
	st.mu.Unlock()

	adbBin, err := resolveAdb()
	if err != nil {
		return ActionResult{OK: false, Message: err.Error()}
	}
	_, serverPath, err := resolveScrcpy()
	if err != nil {
		return ActionResult{OK: false, Message: err.Error()}
	}
	ffmpegBin, err := resolveFFmpeg()
	if err != nil {
		return ActionResult{OK: false, Message: err.Error()}
	}
	adb := &adbClient{bin: adbBin}

	videoLn, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		return ActionResult{OK: false, Message: err.Error()}
	}
	videoPort := videoLn.Addr().(*net.TCPAddr).Port

	httpLn, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		_ = videoLn.Close()
		return ActionResult{OK: false, Message: err.Error()}
	}
	httpPort := httpLn.Addr().(*net.TCPAddr).Port
	streamURL := fmt.Sprintf("http://127.0.0.1:%d/mjpeg", httpPort)

	_, _ = adb.runText(8*time.Second, "-s", deviceID, "reverse", "--remove", "localabstract:scrcpy")
	if _, err := adb.runText(10*time.Second, "-s", deviceID, "reverse", "localabstract:scrcpy", fmt.Sprintf("tcp:%d", videoPort)); err != nil {
		_ = videoLn.Close()
		_ = httpLn.Close()
		return ActionResult{OK: false, Message: "adb reverse 失败: " + err.Error()}
	}
	if _, err := adb.runText(30*time.Second, "-s", deviceID, "push", serverPath, "/data/local/tmp/scrcpy-server.jar"); err != nil {
		_ = videoLn.Close()
		_ = httpLn.Close()
		return ActionResult{OK: false, Message: "push scrcpy-server 失败: " + err.Error()}
	}

	diagnostics := &streamDiagnostics{}
	serverCmd := adb.command("-s", deviceID, "shell",
		"CLASSPATH=/data/local/tmp/scrcpy-server.jar",
		"app_process", "/", "com.genymobile.scrcpy.Server", "5.0.1",
		"tunnel_forward=false",
		"audio=false",
		"control=true",
		"clipboard_autosync=false",
		"cleanup=false",
		"max_size=1280",
		"max_fps=60",
		"video_bit_rate=8000000",
	)
	serverCmd.Stderr = diagnostics
	serverCmd.Stdout = diagnostics
	if err := serverCmd.Start(); err != nil {
		_ = videoLn.Close()
		_ = httpLn.Close()
		return ActionResult{OK: false, Message: "启动 scrcpy-server 失败: " + err.Error()}
	}

	_ = videoLn.(*net.TCPListener).SetDeadline(time.Now().Add(12 * time.Second))
	videoConn, err := videoLn.Accept()
	_ = videoLn.(*net.TCPListener).SetDeadline(time.Time{})
	if err != nil {
		killCmd(serverCmd)
		_ = videoLn.Close()
		_ = httpLn.Close()
		return ActionResult{OK: false, Message: "等待视频连接超时（检查无线调试/本地网络权限）: " + err.Error()}
	}
	// Reverse mode connects video first, then control, before sending metadata.
	_ = videoLn.(*net.TCPListener).SetDeadline(time.Now().Add(12 * time.Second))
	controlConn, err := videoLn.Accept()
	_ = videoLn.Close()
	if err != nil {
		_ = videoConn.Close()
		killCmd(serverCmd)
		_ = httpLn.Close()
		return ActionResult{OK: false, Message: "建立触摸控制通道失败: " + err.Error()}
	}
	installed := false
	defer func() {
		if !installed {
			_ = controlConn.Close()
		}
	}()

	_ = videoConn.SetReadDeadline(time.Now().Add(12 * time.Second))
	if err := readExact(videoConn, make([]byte, 64)); err != nil {
		_ = videoConn.Close()
		killCmd(serverCmd)
		_ = httpLn.Close()
		return ActionResult{OK: false, Message: "读取设备名失败: " + err.Error()}
	}
	if err := readVideoCodec(videoConn); err != nil {
		_ = videoConn.Close()
		killCmd(serverCmd)
		_ = httpLn.Close()
		return ActionResult{OK: false, Message: "读取视频元数据失败: " + err.Error()}
	}

	_ = videoConn.SetReadDeadline(time.Time{})
	ffmpegCmd := exec.Command(ffmpegBin,
		"-hide_banner", "-loglevel", "error",
		"-flags", "low_delay",
		"-probesize", "32",
		"-analyzeduration", "0",
		"-f", "h264",
		"-i", "pipe:0",
		"-an",
		"-fps_mode", "passthrough",
		"-vf", "format=yuvj420p",
		"-f", "image2pipe",
		"-vcodec", "mjpeg",
		"-q:v", "5",
		"pipe:1",
	)
	ffmpegIn, err := ffmpegCmd.StdinPipe()
	if err != nil {
		_ = videoConn.Close()
		killCmd(serverCmd)
		_ = httpLn.Close()
		return ActionResult{OK: false, Message: err.Error()}
	}
	ffmpegOut, err := ffmpegCmd.StdoutPipe()
	if err != nil {
		_ = videoConn.Close()
		killCmd(serverCmd)
		_ = httpLn.Close()
		return ActionResult{OK: false, Message: err.Error()}
	}
	ffmpegCmd.Stderr = diagnostics
	if err := ffmpegCmd.Start(); err != nil {
		_ = videoConn.Close()
		killCmd(serverCmd)
		_ = httpLn.Close()
		return ActionResult{OK: false, Message: "启动 ffmpeg 失败: " + err.Error()}
	}

	stopCh := make(chan struct{})
	st.mu.Lock()
	st.latest.Store([]byte(nil))
	st.stopCh = stopCh
	st.running = true
	st.serial = deviceID
	st.url = streamURL
	st.serverCmd = serverCmd
	st.ffmpegCmd = ffmpegCmd
	st.httpLn = httpLn
	st.videoConn = videoConn
	st.controlConn = controlConn
	st.touchDown = false
	installed = true
	st.mu.Unlock()

	failures := make(chan error, 2)
	st.workers.Add(3)
	go func() { defer st.workers.Done(); _, _ = io.Copy(io.Discard, controlConn) }()
	go func() { defer st.workers.Done(); failures <- st.pumpH264(videoConn, ffmpegIn, stopCh) }()
	go func() {
		defer st.workers.Done()
		st.consumeJPEG(ffmpegOut)
		failures <- fmt.Errorf("视频解码器已结束")
	}()

	mux := http.NewServeMux()
	mux.HandleFunc("/mjpeg", st.handleMJPEG)
	mux.HandleFunc("/snap.jpg", st.handleSnap)
	httpSrv := &http.Server{Handler: mux}
	st.mu.Lock()
	st.httpSrv = httpSrv
	st.mu.Unlock()
	go func() { _ = httpSrv.Serve(httpLn) }()

	deadline := time.Now().Add(8 * time.Second)
	for time.Now().Before(deadline) {
		if b, _ := st.latest.Load().([]byte); len(b) > 100 {
			return ActionResult{OK: true, Message: streamURL}
		}
		select {
		case err := <-failures:
			st.mu.Lock()
			st.stopLocked()
			st.mu.Unlock()
			return ActionResult{OK: false, Message: fmt.Sprintf("预览启动失败: %v %s", err, diagnostics.String())}
		case <-time.After(40 * time.Millisecond):
		}
	}
	st.mu.Lock()
	st.stopLocked()
	st.mu.Unlock()
	return ActionResult{OK: false, Message: "预览启动超时：未收到可解码的视频帧。" + diagnostics.String()}
}

// StopPreviewStream 停止预览流
func (s *Service) StopPreviewStream() ActionResult {
	st := s.ensureStream()
	st.lifecycle.Lock()
	defer st.lifecycle.Unlock()
	st.mu.Lock()
	defer st.mu.Unlock()
	st.stopLocked()
	return ActionResult{OK: true, Message: "已停止预览流"}
}

// GetPreviewStreamURL 当前 MJPEG 地址
func (s *Service) GetPreviewStreamURL() string {
	st := s.ensureStream()
	st.lifecycle.Lock()
	defer st.lifecycle.Unlock()
	st.mu.Lock()
	defer st.mu.Unlock()
	if !st.running {
		return ""
	}
	return st.url
}

func (st *previewStream) stopLocked() {
	if st.stopCh != nil {
		select {
		case <-st.stopCh:
		default:
			close(st.stopCh)
		}
		st.stopCh = nil
	}
	st.running = false
	if st.httpSrv != nil {
		_ = st.httpSrv.Close()
		st.httpSrv = nil
	}
	if st.httpLn != nil {
		_ = st.httpLn.Close()
		st.httpLn = nil
	}
	if st.controlConn != nil {
		if st.touchDown {
			req := st.lastTouch
			req.Action = 3
			_ = st.writeTouchLocked(req)
		}
		_ = st.controlConn.Close()
		st.controlConn = nil
		st.touchDown = false
	}
	if st.videoConn != nil {
		_ = st.videoConn.Close()
		st.videoConn = nil
	}
	if st.videoLn != nil {
		_ = st.videoLn.Close()
		st.videoLn = nil
	}
	killCmd(st.ffmpegCmd)
	killCmd(st.serverCmd)
	st.workers.Wait()
	st.ffmpegCmd = nil
	st.serverCmd = nil
	st.url = ""
	st.serial = ""
	st.clients.Range(func(k, _ any) bool {
		st.clients.Delete(k)
		return true
	})
}

func killCmd(cmd *exec.Cmd) {
	if cmd == nil || cmd.Process == nil {
		return
	}
	_ = cmd.Process.Kill()
	_, _ = cmd.Process.Wait()
}

func readExact(r io.Reader, buf []byte) error {
	_, err := io.ReadFull(r, buf)
	return err
}

// scrcpy 5 sends a 4-byte codec ID, followed by 12-byte session metadata.
// Session headers have no payload; config and video packets do have a payload.
func readVideoCodec(r io.Reader) error {
	var codec [4]byte
	if err := readExact(r, codec[:]); err != nil {
		return err
	}
	if string(codec[:]) != "h264" {
		return fmt.Errorf("不支持的视频编码: %x", codec)
	}
	return nil
}
func (st *previewStream) pumpH264(conn net.Conn, w io.WriteCloser, stop <-chan struct{}) error {
	defer w.Close()
	hdr := make([]byte, 12)
	for {
		select {
		case <-stop:
			return nil
		default:
		}
		// A static phone screen may produce no packets for a long time.
		// Closing the connection in stopLocked interrupts this read when stopping.
		if err := readExact(conn, hdr); err != nil {
			return err
		}
		if binary.BigEndian.Uint64(hdr[:8])&(1<<63) != 0 {
			continue // session header contains flags, width and height, no payload
		}
		size := binary.BigEndian.Uint32(hdr[8:12])
		if size == 0 || size > 16*1024*1024 {
			return fmt.Errorf("无效的视频包长度: %d", size)
		}
		if _, err := io.CopyN(w, conn, int64(size)); err != nil {
			return err
		}
	}
}

// Process output is written asynchronously; retain a bounded, synchronized tail
// so startup errors can be shown to the user without racing the subprocess.
type streamDiagnostics struct {
	mu   sync.Mutex
	data []byte
}

func (d *streamDiagnostics) Write(p []byte) (int, error) {
	d.mu.Lock()
	defer d.mu.Unlock()
	d.data = append(d.data, p...)
	if len(d.data) > 4096 {
		d.data = d.data[len(d.data)-4096:]
	}
	return len(p), nil
}
func (d *streamDiagnostics) String() string {
	d.mu.Lock()
	defer d.mu.Unlock()
	return strings.TrimSpace(string(d.data))
}

func (st *previewStream) consumeJPEG(r io.Reader) {
	br := bufio.NewReaderSize(r, 256*1024)
	window := make([]byte, 0, 128*1024)
	tmp := make([]byte, 64*1024)
	soi := []byte{0xFF, 0xD8}
	eoi := []byte{0xFF, 0xD9}
	for {
		n, err := br.Read(tmp)
		if n > 0 {
			window = append(window, tmp[:n]...)
			for {
				start := bytes.Index(window, soi)
				if start < 0 {
					if len(window) > 2 {
						window = window[len(window)-2:]
					}
					break
				}
				if start > 0 {
					window = window[start:]
				}
				endRel := bytes.Index(window[2:], eoi)
				if endRel < 0 {
					break
				}
				end := endRel + 4
				frame := make([]byte, end)
				copy(frame, window[:end])
				window = window[end:]
				st.latest.Store(frame)
				st.broadcast(frame)
			}
		}
		if err != nil {
			return
		}
		if len(window) > 8*1024*1024 {
			window = window[len(window)/2:]
		}
	}
}

func (st *previewStream) broadcast(frame []byte) {
	st.clients.Range(func(k, _ any) bool {
		ch := k.(chan []byte)
		select {
		case ch <- frame:
		default:
		}
		return true
	})
}

func setCORS(w http.ResponseWriter) {
	w.Header().Set("Access-Control-Allow-Origin", "*")
	w.Header().Set("Access-Control-Allow-Methods", "GET, OPTIONS")
	w.Header().Set("Access-Control-Allow-Headers", "*")
}

func (st *previewStream) handleSnap(w http.ResponseWriter, r *http.Request) {
	setCORS(w)
	if r.Method == http.MethodOptions {
		w.WriteHeader(http.StatusNoContent)
		return
	}
	b, _ := st.latest.Load().([]byte)
	if len(b) == 0 {
		http.Error(w, "no frame", http.StatusServiceUnavailable)
		return
	}
	w.Header().Set("Content-Type", "image/jpeg")
	w.Header().Set("Cache-Control", "no-store")
	_, _ = w.Write(b)
}

func (st *previewStream) handleMJPEG(w http.ResponseWriter, r *http.Request) {
	setCORS(w)
	if r.Method == http.MethodOptions {
		w.WriteHeader(http.StatusNoContent)
		return
	}
	flusher, ok := w.(http.Flusher)
	if !ok {
		http.Error(w, "no flush", http.StatusInternalServerError)
		return
	}
	w.Header().Set("Content-Type", "multipart/x-mixed-replace; boundary=frame")
	w.Header().Set("Cache-Control", "no-cache, no-store, must-revalidate")
	w.Header().Set("Connection", "close")
	flusher.Flush()

	st.mu.Lock()
	stop := st.stopCh
	st.mu.Unlock()
	ch := make(chan []byte, 2)
	st.clients.Store(ch, struct{}{})
	defer st.clients.Delete(ch)

	if b, _ := st.latest.Load().([]byte); len(b) > 0 {
		_ = writeMJPEGPart(w, flusher, b)
	}
	ctx := r.Context()
	for {
		select {
		case <-ctx.Done():
			return
		case <-stop:
			return
		case frame, ok := <-ch:
			if !ok {
				return
			}
			if err := writeMJPEGPart(w, flusher, frame); err != nil {
				return
			}
		}
	}
}

func writeMJPEGPart(w http.ResponseWriter, flusher http.Flusher, frame []byte) error {
	if _, err := fmt.Fprintf(w, "--frame\r\nContent-Type: image/jpeg\r\nContent-Length: %d\r\n\r\n", len(frame)); err != nil {
		return err
	}
	if _, err := w.Write(frame); err != nil {
		return err
	}
	if _, err := w.Write([]byte("\r\n")); err != nil {
		return err
	}
	flusher.Flush()
	return nil
}

func resolveScrcpy() (bin, server string, err error) {
	candidates := []string{"/opt/homebrew/bin/scrcpy", "/usr/local/bin/scrcpy"}
	if p, e := exec.LookPath("scrcpy"); e == nil {
		candidates = append([]string{p}, candidates...)
	}
	var found string
	for _, c := range candidates {
		if fileExists(c) {
			found = c
			break
		}
	}
	serverCandidates := []string{
		"/opt/homebrew/share/scrcpy/scrcpy-server",
		"/usr/local/share/scrcpy/scrcpy-server",
	}
	if cwd, e := os.Getwd(); e == nil {
		serverCandidates = append(serverCandidates, filepath.Join(cwd, "third_party", "scrcpy", "scrcpy-server"))
	}
	if matches, _ := filepath.Glob("/opt/homebrew/Cellar/scrcpy/*/share/scrcpy/scrcpy-server"); len(matches) > 0 {
		serverCandidates = append(serverCandidates, matches...)
	}
	for _, s := range serverCandidates {
		if fileExists(s) {
			if found == "" {
				found = "scrcpy"
			}
			return found, s, nil
		}
	}
	return "", "", fmt.Errorf("未找到 scrcpy-server。请 brew install scrcpy")
}

func resolveFFmpeg() (string, error) {
	candidates := []string{"/opt/homebrew/bin/ffmpeg", "/usr/local/bin/ffmpeg"}
	if p, err := exec.LookPath("ffmpeg"); err == nil {
		candidates = append([]string{p}, candidates...)
	}
	for _, c := range candidates {
		if fileExists(c) {
			return c, nil
		}
	}
	return "", fmt.Errorf("未找到 ffmpeg。请 brew install ffmpeg")
}