608 lines
15 KiB
Go
608 lines
15 KiB
Go
package device
|
||
|
||
import (
|
||
"bufio"
|
||
"bytes"
|
||
"encoding/binary"
|
||
"fmt"
|
||
"io"
|
||
"net"
|
||
"net/http"
|
||
"os/exec"
|
||
"path/filepath"
|
||
"runtime"
|
||
"strings"
|
||
"sync"
|
||
"sync/atomic"
|
||
"time"
|
||
|
||
"github.com/ituoke/ituoke/internal/runtimeassets"
|
||
)
|
||
|
||
// ============================================================
|
||
// [投屏流 · 关键] 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
|
||
screenOff bool
|
||
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, screenOff bool) ActionResult {
|
||
if deviceID == "" {
|
||
s.mu.RLock()
|
||
deviceID = s.active
|
||
s.mu.RUnlock()
|
||
}
|
||
if deviceID == "" {
|
||
return ActionResult{OK: false, Message: "没有选中的设备"}
|
||
}
|
||
s.mu.RLock()
|
||
disconnected := s.disconnectedUSB[deviceID]
|
||
s.mu.RUnlock()
|
||
if disconnected {
|
||
return ActionResult{OK: false, Message: "USB 会话已断开，请先连接设备"}
|
||
}
|
||
|
||
st := s.ensureStream()
|
||
st.lifecycle.Lock()
|
||
defer st.lifecycle.Unlock()
|
||
st.mu.Lock()
|
||
if st.running {
|
||
if st.serial == deviceID {
|
||
if err := st.setScreenOffLocked(screenOff); err != nil {
|
||
st.mu.Unlock()
|
||
return ActionResult{OK: false, Message: err.Error()}
|
||
}
|
||
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 := resolveScrcpyServer()
|
||
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", scrcpyVersion,
|
||
"tunnel_forward=false",
|
||
"audio=false",
|
||
"control=true",
|
||
"clipboard_autosync=false",
|
||
"cleanup=true",
|
||
"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",
|
||
)
|
||
configureToolCommand(ffmpegCmd)
|
||
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 {
|
||
st.mu.Lock()
|
||
err := st.setScreenOffLocked(screenOff)
|
||
if err != nil {
|
||
st.stopLocked()
|
||
}
|
||
st.mu.Unlock()
|
||
if err != nil {
|
||
return ActionResult{OK: false, Message: err.Error()}
|
||
}
|
||
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)
|
||
}
|
||
if st.screenOff {
|
||
_ = st.setScreenOffLocked(false)
|
||
}
|
||
_ = st.controlConn.Close()
|
||
st.controlConn = nil
|
||
st.touchDown = false
|
||
}
|
||
st.screenOff = 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
|
||
}
|
||
|
||
const scrcpyVersion = "5.0.1"
|
||
|
||
func resolveScrcpyServer() (string, error) {
|
||
for _, name := range []string{"scrcpy-server-v" + scrcpyVersion, "scrcpy-server"} {
|
||
if p := runtimeassets.Find(filepath.Join("scrcpy", name)); p != "" {
|
||
return p, nil
|
||
}
|
||
}
|
||
return "", fmt.Errorf("安装包缺少 scrcpy-server %s，请重新安装完整的 iTuoke 安装包", scrcpyVersion)
|
||
}
|
||
|
||
func resolveFFmpeg() (string, error) {
|
||
name := "ffmpeg"
|
||
if runtime.GOOS == "windows" {
|
||
name += ".exe"
|
||
}
|
||
if p := runtimeassets.Find(filepath.Join("ffmpeg", name)); p != "" {
|
||
return p, nil
|
||
}
|
||
// PATH is only a convenience for developers. Release packages include ffmpeg.
|
||
if p, err := exec.LookPath(name); err == nil {
|
||
return p, nil
|
||
}
|
||
return "", fmt.Errorf("安装包缺少 ffmpeg，请重新安装完整的 iTuoke 安装包")
|
||
}
|