Stream 示例

# 离线 smoke
EXAMPLE_MODE=mock go run ./examples/01_quickstart

# 真实链路
EXAMPLE_MODE=real \
INPUT_RTSP_URI="rtsp://camera/live" \
go run -tags gogst ./examples/01_quickstart

# Mock 模式
EXAMPLE_MODE=mock go run ./examples/01_quickstart

# Real 模式
EXAMPLE_MODE=real \
INPUT_RTSP_URI="rtsp://camera/live" \
go run -tags gogst ./examples/01_quickstart

// 创建 GStreamer 引擎
engine := stream.NewGoGSTEngine()

// 创建管理器（需要 Engine、Source、Sink 中的至少一个）
manager := stream.NewManager(engine, nil, nil)
defer manager.Close()

// 创建一键调用 API
api := stream.NewOneCallAPI(manager, stream.OneCallConfig{})

// 构建请求
req := stream.OneCallRequest{
    ID:         "demo-01",
    URI:        "rtsp://camera/live",
    SourceType: stream.SourceTypeRTSP,
}

// 启动会话
result := api.StartSession(req)
if result.Error != nil {
    panic(fmt.Sprintf("启动失败: %v", result.Error))
}

// 等待观察
time.Sleep(5 * time.Second)

// 停止会话
api.StopSession("demo-01")

EXAMPLE_MODE=mock go run ./examples/02_preset_override

result := api.StartSession(stream.OneCallRequest{
    ID:         "preset-01",
    URI:        "rtsp://camera/live",
    SourceType: stream.SourceTypeRTSP,
    Preset:     stream.PresetStorageArchiveH265Compact,
    Override: stream.ProfileOverride{
        Transcode: &stream.TranscodeSpec{
            VideoCodec: "h265",
            Bitrate:    1000000, // 显式覆盖码率
        },
    },
})

EXAMPLE_MODE=mock go run ./examples/03_governance_checkpoint

// 设置鉴权器
manager.SetAuthorizer(stream.AuthorizerFunc(func(ctx context.Context, op string, spec stream.SessionSpec) error {
    if !strings.HasPrefix(spec.ID, "allowed-") {
        return fmt.Errorf("会话 ID 必须以 'allowed-' 开头")
    }
    return nil
}))

// 设置审计器
manager.SetAuditor(stream.AuditorFunc(func(ctx context.Context, op string, spec stream.SessionSpec, err error) {
    log.Printf("[AUDIT] op=%s id=%s err=%v", op, spec.ID, err)
}))

// 设置检查点存储
manager.SetCheckpointStore(stream.NewMemoryCheckpointStore())

// 启动会话
session, err := manager.StartSession(ctx, spec)

// 进程重启后恢复
recovered, err := manager.RecoverSessions(ctx)

EXAMPLE_MODE=mock go run ./examples/04_fusion_fanout

spec := stream.SessionSpec{
    ID: "fusion-demo",
    Pipeline: stream.PipelineSpec{
        Pull: stream.SourceSpec{
            ID:   "camera",
            Type: stream.SourceTypeRTSP,
            URI:  "rtsp://camera/live",
        },
    },
    Fusion: stream.FusionSpec{
        Enabled: true,
        FanOut: []stream.FusionFanOutTarget{
            {ID: "rtmp", URI: "rtmp://server/live/stream"},
            {ID: "file", URI: "file:///tmp/record.mp4"},
        },
    },
}

# 预览模式（默认）
go run ./examples/05_rtsp_to_rtmp_template

# 真实模式
REAL_MODE=true \
INPUT_RTSP_URI="rtsp://camera/live" \
OUTPUT_URI="rtmp://media/live/stream" \
go run -tags gogst ./examples/05_rtsp_to_rtmp_template

req := stream.VideoUnitRequest{
    EncodeType: stream.VideoUnitEncodeH264,
    OutputType: stream.VideoUnitOutputRTMP,
    OutputURI:  "rtmp://media/live/stream",
    RTSPURL:    "rtsp://camera/live",
    TaskID:     "rtmp-push-01",
    Width:      1920,
    Height:     1080,
    FPS:        25,
    LatencyMS:  200,
    X26xExtra: stream.VideoUnitX26xExtra{
        BPS:    2500 * 1000,
        GOP:    50,
        QPInit: 26,
        QPMin:  26,
        QPMax:  50,
        RCMode: 0, // CBR
    },
}

EXAMPLE_MODE=mock go run ./examples/06_persistent_checkpoint

// 创建文件检查点存储
store := stream.NewFileCheckpointStore("/tmp/stream_checkpoints_example.json")
manager.SetCheckpointStore(store)

// 启动会话
session, err := manager.StartSession(ctx, spec)

// 模拟进程重启
manager.Close()

// 创建新管理器并恢复
newManager := stream.NewManager(engine, nil, nil)
newManager.SetCheckpointStore(store)
recovered, err := newManager.RecoverSessions(ctx)

bash ./examples/07_real_gst_pipeline_template/check_gst_env.sh

GST_ROOT="/workspace/app/dynamic/1.28.1" \
PKG_CONFIG_PATH="$GST_ROOT/lib/pkgconfig" \
CGO_CFLAGS="-I$GST_ROOT/include" \
CGO_LDFLAGS="-L$GST_ROOT/lib" \
LD_LIBRARY_PATH="$GST_ROOT/lib:$LD_LIBRARY_PATH" \
REAL_MODE=true \
INPUT_RTSP_URI="rtsp://camera/live" \
OUTPUT_URI="rtmp://media/live/stream" \
go run -tags gogst ./examples/07_real_gst_pipeline_template

// 使用 VideoUnit facade 封装 GStreamer 管道
result := stream.ValidateVideoUnitRequest(ctx, req)
if !result.Accepted {
    log.Printf("配置验证失败: %s, 建议: %s", result.ErrorCode, result.NextAction)
    return
}

unit, err := stream.NewVideoUnit(ctx, req, engine, nil)
if err != nil {
    log.Printf("创建 VideoUnit 失败: %v", err)
    return
}

// 启动处理
if err := unit.Start(); err != nil {
    log.Printf("启动失败: %v", err)
    return
}

# Mock 模式
EXAMPLE_MODE=mock go run ./examples/08_multi_session_scheduler

# Real 模式
EXAMPLE_MODE=real \
SCHED_INPUT_RTSP_TEMPLATE="rtsp://camera-%02d/live" \
SCHED_OUTPUT_URI_TEMPLATE="rtmp://media/live/scheduler-%02d" \
go run -tags gogst ./examples/08_multi_session_scheduler

scheduler := stream.NewSessionStartScheduler(manager, stream.SessionStartSchedulerConfig{
    MaxConcurrentStarts: 2,
    StartThrottle:       150 * time.Millisecond,
    RetryMax:            4,
})

for i := 0; i < sessionCount; i++ {
    id := fmt.Sprintf("sched-%03d", i)
    req := stream.OneCallRequest{
        ID:         id,
        URI:        fmt.Sprintf(inputTemplate, i),
        SourceType: stream.SourceTypeRTSP,
    }
    scheduler.Schedule(ctx, id, req)
}

// 等待所有会话启动完成
results := scheduler.Wait(ctx)

EXAMPLE_MODE=real \
INPUT_RTSP_URI="rtsp://camera/live" \
OUTPUT_URI="rtmp://media/live/stream" \
go run -tags gogst ./examples/09_alerting_exporter

// 订阅事件
events.Subscribe("alerter", stream.EventHandlerFunc(func(e stream.Event) {
    switch e.Type {
    case stream.EventReconnect:
        reconnectCount.Add(1)
        if reconnectCount.Load() >= alertReconnectThreshold {
            triggerAlert("重连次数过多")
        }
    case stream.EventError:
        if e.Severity == stream.SeverityCritical {
            triggerAlert("严重错误")
        }
    case stream.EventHealthScore:
        if score, ok := e.Data["score"].(int); ok && score < 50 {
            triggerAlert("健康分过低")
        }
    }
}))

EXAMPLE_MODE=mock go run ./examples/10_blue_green_rebuild

// 获取当前规格
currentSpec := session.SpecSnapshot()

// 修改配置
newSpec := currentSpec
newSpec.Pipeline.Transcode.Bitrate = 5000000 // 提高码率

// 执行蓝绿重建
err := session.Rebuild(ctx, newSpec)
if err != nil {
    // 重建失败，自动回滚到原配置
    log.Printf("重建失败，回滚: %v", err)
    session.Rebuild(ctx, currentSpec)
}

EXAMPLE_MODE=mock go run ./examples/11_disaster_recovery_drill

EXAMPLE_MODE=mock go run ./examples/12_cloud_uplink_profiles

go run ./examples/13_hardware_diagnostics

// 执行硬件诊断
diagnostics := api.DiagnoseHardware()

// 打印能力快照
fmt.Printf("可用厂商: %v\n", diagnostics.Snapshot.AvailableVendors)
fmt.Printf("可用加速器: %v\n", diagnostics.Snapshot.AvailableAccelerators)

// 打印缺失能力
for _, vendor := range diagnostics.Vendors {
    if len(vendor.MissingEncoders) > 0 {
        fmt.Printf("厂商 %s 缺失编码器: %v\n", vendor.Vendor, vendor.MissingEncoders)
    }
}

# 预览
go run ./examples/14_news_pip_composite

# 真实链路
EXAMPLE_MODE=real \
MAIN_RTSP_URI="rtsp://cam-main/live" \
PIP_RTSP_URI="rtsp://cam-guest/live" \
OUTPUT_URI="rtmp://media/live/news-pip" \
go run -tags gogst ./examples/14_news_pip_composite

spec := stream.SessionSpec{
    ID: "news-pip",
    Pipeline: stream.PipelineSpec{
        Composite: stream.CompositeSpec{
            Enabled:    true,
            Mode:       stream.CompositeModePIP,
            Width:      1280,
            Height:     720,
            FPS:        15,
            Background: "black",
            AudioMode:  stream.CompositeAudioFirst,
            Inputs: []stream.CompositeInputSpec{
                {
                    ID: "main", Type: stream.SourceTypeRTSP,
                    URI: mainRTSP,
                    X: 0, Y: 0, Width: 1280, Height: 720, ZOrder: 0,
                },
                {
                    ID: "pip", Type: stream.SourceTypeRTSP,
                    URI: pipRTSP,
                    X: 1056, Y: 606, Width: 200, Height: 112, ZOrder: 10,
                },
            },
        },
    },
}

go run ./examples/22_timeseries_aggregation

// 配置时间序列聚合器
cfg := stream.TimeSeriesConfig{
    Enabled:        true,
    SampleInterval: 30 * time.Second,
    MaxAge:         7 * 24 * time.Hour,
}
manager.SetTimeSeriesConfig(ctx, cfg)

// 查询历史统计
end := time.Now()
start := end.Add(-1 * time.Hour)
result, _ := manager.QueryTimeRange("session-id", start, end)

fmt.Printf("平均码率: %.2f Kbps\n", result.AvgBitrateKbps)
fmt.Printf("平均健康分: %.2f\n", result.AvgHealthScore)

# 预设多源模式（高资源，低延迟，支持淡入淡出）
SWITCH_MODE=preset \
OUTPUT_URI=rtmp://media/live/news \
AUTO_SWITCH=true \
AUTO_SWITCH_INTERVAL=10s \
go run -tags gogst ./examples/23_seamless_switch

# 动态切换模式（低资源，短暂黑屏）
SWITCH_MODE=dynamic \
OUTPUT_URI=rtmp://media/live/news \
AUTO_SWITCH=true \
go run -tags gogst ./examples/23_seamless_switch

// 预设模式：所有源同时拉流，通过 Alpha 切换
Inputs: []CompositeInputSpec{
    {ID: "anchor", Alpha: 1.0, ZOrder: 0},  // 显示主播
    {ID: "field1", Alpha: 0.0, ZOrder: 1},  // 隐藏现场1
    {ID: "field2", Alpha: 0.0, ZOrder: 2},  // 隐藏现场2
}

// 淡入淡出切换
for step := 0; step <= 10; step++ {
    alpha := float64(step) / 10.0
    newComposite.Inputs[0].Alpha = 1.0 - alpha  // 主播淡出
    newComposite.Inputs[1].Alpha = alpha        // 现场1淡入
    manager.UpdateCompositeLayout(ctx, sessionID, newComposite)
    time.Sleep(50 * time.Millisecond)
}

// 动态模式：只拉流当前源，切换时更新 URI
newComposite.Inputs[0].URI = "rtsp://field1/live"
manager.UpdateCompositeLayout(ctx, sessionID, newComposite)

# 检查 GStreamer 版本
gst-launch-1.0 --version

# 检查可用插件
gst-inspect-1.0 | grep -E "(rtsp|rtmp|h264)"

# 测试管道
gst-launch-1.0 uridecodebin uri="rtsp://..." ! ... ! fakesink

# 检查 CGO
go env CGO_ENABLED

# 检查 pkg-config
pkg-config --modversion gstreamer-1.0

# 清理重建
go clean -cache && go build -tags gogst ./...