stream/lowlatency

┌─────────────────────────────────────────────────────────────┐
│                 LowLatencyPipeline                          │
│  ┌─────────────┐  ┌─────────────────┐  ┌─────────────────┐ │
│  │ Transport   │  │ LatencyOptimizer│  │ JitterBuffer    │ │
│  │ (SRT/WebRTC │  │                 │  │                 │ │
│  │  /RTMP)     │  │ measureLatency  │  │ avg + 2σ        │ │
│  │             │  │  → Optimize     │  │  → targetDelay  │ │
│  └──────┬──────┘  └────────┬────────┘  └────────┬────────┘ │
│         │                  │                    │          │
│         ▼                  ▼                    ▼          │
│  BuildPipeline()    ApplyTargetLatency()   Push/PopFrame   │
└─────────────────────────────────────────────────────────────┘

targetDelay = avgInterval + 2 * stdDev   // 覆盖 95% 样本
targetDelay = clamp(targetDelay, MinDelay, MaxDelay)
targetDelay = smooth(targetDelay, old, weight=0.7)  // 加权平滑

import (
    "cnb.cool/svn/stream"
    "cnb.cool/svn/stream/lowlatency"
)

events := stream.NewEventBus(stream.EventBusConfig{})
defer events.Close()

pipeline := lowlatency.NewLowLatencyPipeline(
    lowlatency.LowLatencyConfig{
        TransportType: "srt",
        SRTConfig: lowlatency.SRTConfig{
            Latency:      100,  // 100ms
            MaxBandwidth: 5000, // 5Mbps
            OverheadBW:   25,   // 25% 开销
            StreamID:     "stream-01",
            Passphrase:   "your-password",
            KeySize:      128,  // AES-128
        },
        LatencyOptimizer: lowlatency.LatencyOptimizeConfig{
            TargetLatency:  100 * time.Millisecond,
            MinLatency:     20 * time.Millisecond,
            MaxLatency:     500 * time.Millisecond,
            EnableAdaptive: true,
        },
        JitterBuffer: lowlatency.JitterBufferConfig{
            InitialDelay: 50 * time.Millisecond,
            AdaptiveMode: true,
            WindowSize:   32,
        },
    },
    session,      // *stream.Session（满足 sessionProvider 接口）
    healthTracker, // *stream.HealthTracker
    events,
)

// 构建 GStreamer Pipeline 字符串
p, err := pipeline.BuildPipeline("srt://example.com:9000")

pipeline := lowlatency.NewLowLatencyPipeline(
    lowlatency.LowLatencyConfig{
        TransportType: "webrtc",
        WebRTCConfig: lowlatency.WebRTCConfig{
            STUNServers:        []string{"stun:stun.l.google.com:19302"},
            LatencyMode:        lowlatency.WebRTCLatencyLow,
            Bandwidth:          2000, // 2Mbps
            BundlePolicy:       "max-bundle",
            ICETransportPolicy: "all",
        },
        LatencyOptimizer: lowlatency.LatencyOptimizeConfig{
            TargetLatency:  80 * time.Millisecond,
            EnableAdaptive: true,
        },
    },
    session, healthTracker, events,
)

p, err := pipeline.BuildPipeline("")

ctx, cancel := context.WithCancel(context.Background())
defer cancel()

// 启动自动优化（阻塞，需 goroutine）
go pipeline.Run(ctx)

// 手动触发一次优化
if err := pipeline.Optimize(ctx); err != nil {
    log.Printf("优化失败: %v", err)
}

// 推送帧
if err := pipeline.PushFrame(frame); err != nil {
    log.Printf("入缓冲失败: %v", err)
}

// 按目标延迟弹出帧（可能返回 nil，表示未到播放时间）
frame := pipeline.PopFrame()
if frame != nil {
    // 处理帧
}

snap := pipeline.Snapshot()
fmt.Printf("传输: %s, 应用延迟: %v, 当前延迟: %v, 缓冲大小: %d\n",
    snap.TransportType,
    snap.AppliedLatency,
    snap.CurrentLatency,
    snap.JitterBufferSize,
)

// SRT 指标
stats := pipeline.GetTransportStats().(lowlatency.SRTStats)
fmt.Printf("配置延迟: %d, 丢包: %d, RTT: %d\n",
    stats.ConfiguredLatency, stats.PacketsLost, stats.RoundTripDelay)

// WebRTC 指标
stats := pipeline.GetTransportStats().(lowlatency.WebRTCStats)
fmt.Printf("可用带宽: %d, RTT: %d, 丢帧: %d\n",
    stats.AvailableBitrate, stats.RoundTripDelay, stats.FramesDropped)

// 全局指标
snap := pipeline.Snapshot()
fmt.Printf("缓冲大小: %d, 当前延迟: %v\n",
    snap.JitterBufferSize, snap.CurrentLatency)

// SRT 加密
SRTConfig: lowlatency.SRTConfig{
    Passphrase: "your-strong-password",
    KeySize:    256,  // AES-256
}

// WebRTC TURN 认证
TURNServers: []lowlatency.TURNServer{
    {
        URL:      "turn:turn.example.com:3478",
        Username: "user",
        Password: "pass",
    },
}

# 运行所有测试
cd /workspace
go test -v ./lowlatency/...

# 运行短测试（跳过耗时测试）
go test -v -short ./lowlatency/...

# 运行基准测试
go test -bench=. ./lowlatency/...

lowlatency/
├── DESIGN.md                    # 架构设计文档
├── README.md                    # 本文档
├── lowlatency_pipeline.go       # 主编排器（传输 + 优化 + 缓冲）
├── lowlatency_pipeline_test.go  # Pipeline 集成测试
├── srt_transport.go             # SRT 传输协议
├── srt_transport_test.go        # SRT 单元测试
├── webrtc_transport.go          # WebRTC 传输协议
├── webrtc_transport_test.go     # WebRTC 单元测试
├── latency_optimizer.go         # 自适应延迟优化器
├── latency_optimizer_test.go    # 优化器单元测试
├── jitter_buffer.go             # 抖动缓冲 + 环形缓冲
└── jitter_buffer_test.go        # 抖动缓冲单元测试