如何为传统遗留相机创建响应式Observable封装

核心实现代码 #

using System;
using System.Reactive;
using System.Reactive.Linq;

public class ObservableCamera : IDisposable
{
    private readonly IDisposable _globalCleanup;
    public IObservable<Frame[]> Captures { get; }

    public ObservableCamera(Camera camera, IObservable<bool> enableCapture)
    {
        // 1. 将原生OnCapture事件转换为标准帧流
        var frameStream = Observable.FromEventPattern<CaptureEventHandler, Frame>(
            conversion: handler => (sender, capture) => handler(capture),
            addHandler: handler => camera.OnCapture += handler,
            removeHandler: handler => camera.OnCapture -= handler
        ).Select(pattern => pattern.EventArgs);

        // 2. 预处理控制流：过滤重复状态、转为热共享流
        var hotControl = enableCapture
            .DistinctUntilChanged()
            .Publish()
            .RefCount();

        // 3. 构建捕获序列
        Captures = hotControl
            // 仅在切换为启用状态时触发新捕获周期
            .Where(isEnabled => isEnabled)
            .SelectMany(_ => 
                // 自动管理相机启停资源：创建时启动，释放时必然停止
                Observable.Using(
                    resourceFactory: () => new CaptureScope(camera),
                    // 基于资源构建当前周期的帧缓冲流
                    streamFactory: _ => frameStream
                        // 捕获周期停止条件：当前启用时段内首次收到false，或控制流终止
                        .TakeUntil(
                            stopSignal: hotControl
                                // 跳过当前刚触发的true信号，监听后续状态
                                .Skip(1)
                                .Where(isEnabled => !isEnabled)
                                .Take(1)
                        )
                        // 整个捕获周期的帧打包为数组，周期结束时推送
                        .Buffer()
                )
            )
            // 多订阅共享同一份捕获逻辑，避免重复启停相机
            .Publish()
            .RefCount();

        // 全局清理句柄
        _globalCleanup = Captures.Subscribe(_ => { }, _ => { }, () => { });
    }

    // 内部资源类：封装相机启停逻辑，保证Dispose时必然执行停止
    private class CaptureScope : IDisposable
    {
        private readonly Camera _camera;
        public CaptureScope(Camera camera)
        {
            _camera = camera;
            _camera.StartRecording();
        }

        public void Dispose()
        {
            try
            {
                _camera.StopRecording();
            }
            catch
            {
                // 吞掉停止阶段的非致命异常，避免资源清理流程崩溃
            }
        }
    }

    // 外部主动调用清理全流程
    public void Dispose()
    {
        _globalCleanup?.Dispose();
    }
}

核心逻辑说明 #

• 事件适配：通过Observable.FromEventPattern将原生OnCapture事件转换为标准IObservable<Frame>流，自动处理事件的订阅与退订，避免内存泄漏。
• 控制流预处理：对enableCapture执行DistinctUntilChanged，过滤连续重复的同状态值，避免重复调用Start/Stop引发异常；通过Publish().RefCount()将控制流转为热共享流，供内部多个逻辑节点订阅时不会重复拉取原始控制流数据。
• 资源自动管控：通过Observable.Using将「启动录制、停止录制」封装为IDisposable资源，无论捕获周期正常结束、触发停止、下游取消订阅还是流程抛出异常，资源释放逻辑（调用StopRecording）都会被强制执行，完全满足约束要求。
• 捕获周期控制：仅当控制流推送true时触发新的捕获周期；每个周期内的帧流通过TakeUntil监听后续的false信号或控制流终止信号，周期结束时自动停止录制。
• 批量输出：通过Buffer()操作符将单个捕获周期内收集到的所有帧聚合，周期结束时统一打包为Frame[]推送，匹配输出序列的类型要求。

边界场景适配 #

• 若enableCapture推送true后直接终止（未推送false）：当前捕获周期随控制流终止自动结束，自动调用StopRecording，并将已捕获的帧打包推送。
• 若下游订阅Captures后主动取消订阅：当前捕获周期立即终止，自动调用StopRecording。
• 若enableCapture推送异常：异常直接传递给下游订阅者，已启动的捕获周期自动终止，StopRecording仍会被执行。
• 若连续推送多个相同的true/false：DistinctUntilChanged会自动过滤重复值，不会重复触发相机启停逻辑。

使用示例 #

// 构建控制流：第2秒开始录制，第12秒停止
var enableFlow = Observable.Timer(TimeSpan.FromSeconds(2))
    .Select(_ => true)
    .Concat(Observable.Timer(TimeSpan.FromSeconds(10)).Select(_ => false));

// 初始化实例
var nativeCamera = new Camera();
var obsCamera = new ObservableCamera(nativeCamera, enableFlow);

// 订阅批量帧输出
obsCamera.Captures.Subscribe(
    frames => Console.WriteLine($"本次捕获共{frames.Length}帧"),
    ex => Console.WriteLine($"捕获出错: {ex.Message}"),
    () => Console.WriteLine("全流程捕获结束")
);

内容的提问来源于stack exchange，提问作者SuperJMN