using Serilog; namespace SHH.CameraSdk; /// /// [架构基类] 工业级视频源抽象核心 (V3.3.4 严格匹配版) /// 核心职责: /// 1. 提供线程安全的生命周期管理(启动/停止/销毁) /// 2. 实现状态变更的可靠分发与异常隔离 /// 3. 支持配置热更新与动态参数应用 /// 4. 内置 FPS/码率统计、心跳保活、审计日志能力 /// 关键修复记录: /// ✅ [Bug A] 死锁免疫:所有 await 均添加 ConfigureAwait(false),解除 UI 线程依赖 /// ✅ [Bug π] 管道安全:Dispose 采用优雅关闭策略,确保剩余状态消息被消费 /// ✅ [编译修复] 补全 CloneConfig 中 Transport/VendorArguments 的深拷贝逻辑 /// public abstract class BaseVideoSource : IVideoSource, IAsyncDisposable, IDeviceConnectivity { // [新增] 物理在线状态(专门给 Ping 使用) private volatile bool _isPhysicalOnline; public bool IsPhysicalOnline => _isPhysicalOnline; /// /// 图像预处理配置(缩放、增量等) /// 放置在基类中,确保所有接入协议(HIK/DH/RTSP)均可共享处理逻辑 /// public PreprocessConfig PreprocessSettings { get; set; } = new PreprocessConfig(); string IDeviceConnectivity.IpAddress => _config.IpAddress; // 允许哨兵从外部更新 _isOnline 字段 void IDeviceConnectivity.SetNetworkStatus(bool isOnline) { if (_isPhysicalOnline != isOnline) { _isPhysicalOnline = isOnline; // 触发状态变更是为了通知 UI 更新绿色小圆点,但不改变 Status // 注意:这里传 _status 保持原样,只变消息 StatusChanged?.Invoke(this, new StatusChangedEventArgs(_status, isOnline ? "物理网络恢复" : "物理网络中断")); } } protected abstract ILogger _sdkLog { get; } #region --- 1. 核心配置与锁机制 (Core Config & Locks) --- /// /// 核心配置对象(支持热更新,去除 readonly 修饰符) /// 注意:外部修改需通过 UpdateConfig 方法,确保线程安全 /// protected VideoSourceConfig _config; /// /// 状态同步锁 /// 作用:保护 _status 字段的读写原子性,防止多线程竞争导致状态不一致 /// private readonly object _stateSyncRoot = new(); /// /// 生命周期互斥锁(信号量) /// 作用:确保 StartAsync/StopAsync/UpdateConfig 串行执行,防止重入导致状态机混乱 /// 配置:初始计数 1,最大计数 1 → 互斥锁 /// private readonly SemaphoreSlim _lifecycleLock = new(1, 1); #endregion #region --- 2. 内部状态与基础设施 (Internal States & Infrastructure) --- /// 设备在线状态标志(volatile 确保多线程可见性) private volatile bool _isOnline; /// 视频源核心状态(受 _stateSyncRoot 保护) private VideoSourceStatus _status = VideoSourceStatus.Disconnected; /// /// 状态通知有界通道 /// 特性:DropOldest 策略,消费者过载时丢弃旧状态,防止内存溢出 /// 配置:容量 100 | 单读者多写者 /// private readonly Channel _statusQueue; /// 状态分发器的取消令牌源 private CancellationTokenSource? _distributorCts; /// 状态分发任务引用(用于 Dispose 时优雅等待) private Task? _distributorTask; /// 最后一帧接收的系统 Tick(单调时钟,不受系统时间修改影响) private long _lastFrameTick = 0; /// 视频帧回调事件(热路径,低延迟分发) public event Action? FrameReceived; #endregion #region --- 3. 公开属性 (Public Properties) --- /// 视频源唯一标识 public long Id => _config.Id; /// 只读配置副本(外部仅能通过 UpdateConfig 修改) public VideoSourceConfig Config => _config; /// 视频源当前状态(线程安全读取) public VideoSourceStatus Status { get { lock (_stateSyncRoot) { return _status; } } } /// 运行状态标记 public bool IsRunning { get; set; } /// 设备在线状态 public bool IsOnline => _isOnline; /// 设备元数据(能力集、通道信息等) public DeviceMetadata Metadata { get; protected set; } = new(); /// 状态变更事件(对外暴露状态通知) public event EventHandler? StatusChanged; /// 最后一帧接收的 Tick 时间戳(线程安全读取) public long LastFrameTick => Interlocked.Read(ref _lastFrameTick); #endregion #region --- 4. 遥测统计属性 (Telemetry Properties) --- /// 生命周期内接收的总帧数 private long _totalFramesReceived = 0; /// FPS 计算临时计数器 private int _tempFrameCounter = 0; /// 上次 FPS 计算的 Tick 时间 private long _lastFpsCalcTick = 0; // 提供一个最近一秒的输入帧率参考值 public int NominalInputFps => (int)Math.Round(RealFps); /// 实时 FPS(每秒更新一次) public double RealFps { get; private set; } = 0.0; /// 实时码率 (Mbps) protected double _currentBitrate = 0; public double RealBitrate => _currentBitrate; /// 码率计算临时字节计数器 private long _tempByteCounter = 0; /// 生命周期总帧数(线程安全读取) public long TotalFrames => Interlocked.Read(ref _totalFramesReceived); #endregion #region --- 5. 审计日志系统 (Audit Log System) --- /// 审计日志列表(线程安全访问) private readonly List _auditLogs = new(); /// 最大日志条数(滚动清除,防止内存溢出) private const int MaxAuditLogCount = 100; #endregion #region --- 6. 构造函数 (Constructor) --- /// /// 初始化视频源基础设施 /// /// 视频源基础配置 /// 配置为空时抛出 protected BaseVideoSource(VideoSourceConfig config) { // 入参校验 _config = config ?? throw new ArgumentNullException(nameof(config), "视频源配置不能为空"); // 初始化帧控制器 Controller = new FrameController(); // 配置深拷贝(防漂移:内部配置与外部引用隔离) _config = CloneConfig(config); // 初始化有界状态通道 _statusQueue = Channel.CreateBounded(new BoundedChannelOptions(100) { FullMode = BoundedChannelFullMode.DropOldest, SingleReader = true, SingleWriter = false }); // 初始化状态分发器 _distributorCts = new CancellationTokenSource(); _distributorTask = Task.Run(() => StatusDistributorLoopAsync(_distributorCts.Token)); } #endregion #region --- 7. 配置管理 (Config Management) --- /// /// 热更新视频源配置(线程安全) /// 新配置将在下次启动/重连时生效 /// /// 新的视频源配置 public void UpdateConfig(VideoSourceConfig newConfig) { if (newConfig == null) return; // 加生命周期锁:防止与启动/停止操作并发 _lifecycleLock.Wait(); try { // 深拷贝新配置,隔离外部引用 _config = newConfig.DeepCopy(); // 写入审计日志 AddAuditLog($"配置已更新 [IP:{_config.IpAddress}],生效时机:{(_isOnline ? "下次重连" : "下次启动")}"); Debug.WriteLine($"[ConfigUpdated] 设备 {Id} 配置落地完成"); } finally { _lifecycleLock.Release(); } } /// /// 配置深拷贝辅助方法(确保引用类型独立) /// /// 源配置 /// 深拷贝后的新配置 private VideoSourceConfig CloneConfig(VideoSourceConfig source) { return new VideoSourceConfig { Id = source.Id, Brand = source.Brand, IpAddress = source.IpAddress, Port = source.Port, Username = source.Username, Password = source.Password, ChannelIndex = source.ChannelIndex, StreamType = source.StreamType, Transport = source.Transport, ConnectionTimeoutMs = source.ConnectionTimeoutMs, MainboardIp = source.MainboardIp, MainboardPort = source.MainboardPort, RtspPath = source.RtspPath, RenderHandle = source.RenderHandle, // Dictionary 深拷贝:防止外部修改影响内部 VendorArguments = source.VendorArguments != null ? new Dictionary(source.VendorArguments) : new Dictionary() }; } #endregion #region --- 8. 生命周期控制 (Lifecycle Control) --- /// /// 异步启动设备连接 /// 包含:状态校验、非托管初始化、元数据刷新 /// public async Task StartAsync() { // 死锁免疫:不捕获当前同步上下文 await _lifecycleLock.WaitAsync().ConfigureAwait(false); try { // 等待上一个生命周期任务完成 await _pendingLifecycleTask.ConfigureAwait(false); // 幂等性检查:已在线则直接返回 if (_isOnline) return; // 更新状态为连接中 UpdateStatus(VideoSourceStatus.Connecting, $"正在启动 {_config.Brand} 设备..."); // 驱动层启动逻辑(带 15 秒超时) using var cts = new CancellationTokenSource(TimeSpan.FromSeconds(15)); await OnStartAsync(cts.Token).ConfigureAwait(false); // ============================================================ // 【核心修复位置】 // 赋予心跳 5 秒的宽限期 (Grace Period) // 这样 Coordinator 在未来 5 秒内计算出来的无帧时间将是负数或极小值,不会触发复位 // ============================================================ Interlocked.Exchange(ref _lastFrameTick, Environment.TickCount64 + 5000); // 标记运行状态 _isOnline = true; IsRunning = true; // 更新状态为播放中,并刷新元数据 UpdateStatus(VideoSourceStatus.Playing, "流传输正常运行"); await RefreshMetadataAsync().ConfigureAwait(false); } catch (Exception ex) { // 异常回滚:标记离线并更新状态 _isOnline = false; UpdateStatus(VideoSourceStatus.Disconnected, $"启动失败: {ex.Message}"); throw; // 向上抛出异常,由上层处理 } finally { _lifecycleLock.Release(); } } /// /// 异步停止设备连接 /// public async Task StopAsync() { await _lifecycleLock.WaitAsync().ConfigureAwait(false); try { // 标记离线,阻断后续数据处理 _isOnline = false; // 执行驱动层停止逻辑 await OnStopAsync().ConfigureAwait(false); } finally { // 更新状态并释放锁 UpdateStatus(VideoSourceStatus.Disconnected, "连接已手动断开"); _lifecycleLock.Release(); } } /// /// 刷新设备元数据,对比差异并更新 /// /// 元数据差异描述符 public async Task RefreshMetadataAsync() { // 离线状态不刷新元数据 if (!_isOnline) return MetadataDiff.None; try { // 获取驱动层最新元数据 var latestMetadata = await OnFetchMetadataAsync().ConfigureAwait(false); if (latestMetadata != null && latestMetadata.ChannelCount > 0) { // 对比新旧元数据差异 var diff = Metadata.CompareWith(latestMetadata); // 更新元数据并标记同步时间 Metadata = latestMetadata; Metadata.MarkSynced(); return diff; } } catch (Exception ex) { AddAuditLog($"元数据刷新失败: {ex.Message}"); Debug.WriteLine($"[MetadataWarning] 设备 {Id}: {ex.Message}"); } return MetadataDiff.None; } /// /// 应用动态流参数(运行时热更新) /// 支持:分辨率切换、码流类型切换、渲染句柄变更等 /// /// 动态配置项 public void ApplyOptions(DynamicStreamOptions options) { // 离线或参数为空时,忽略请求 if (options == null || !_isOnline) { AddAuditLog("动态参数应用失败:设备离线或参数为空"); return; } try { // 1. 基于设备能力校验参数合法性 if (Metadata.ValidateOptions(options, out string errorMsg)) { // 2. 执行驱动层参数应用逻辑 OnApplyOptions(options); UpdateStatus(_status, "动态参数已成功应用"); // 3. 记录参数变更日志 var changeLog = new List(); if (options.StreamType.HasValue) changeLog.Add($"码流类型={options.StreamType}"); if (options.RenderHandle.HasValue) changeLog.Add($"渲染句柄已更新"); if (options.TargetAnalyzeFps.HasValue) changeLog.Add($"分析帧率={options.TargetAnalyzeFps}fps"); AddAuditLog($"动态参数应用: {string.Join(" | ", changeLog)}"); } else { Debug.WriteLine($"[OptionRejected] 设备 {Id}: {errorMsg}"); } } catch (Exception ex) { AddAuditLog($"动态参数应用失败: {ex.Message}"); Debug.WriteLine($"[ApplyOptionsError] {ex.Message}"); } } /// /// 驱动层参数应用逻辑(子类重写) /// /// 动态配置项 protected virtual void OnApplyOptions(DynamicStreamOptions options) { } #endregion #region --- 9. 帧处理与状态管理 (Frame Processing & Status Management) --- /// /// 检查是否存在帧订阅者(性能优化) /// 无订阅时可跳过解码/预处理等耗时操作 /// /// 有订阅者返回 true protected bool HasFrameSubscribers() => FrameReceived != null; /// /// 上报驱动层异常,触发重连自愈逻辑 /// /// 相机统一异常 protected void ReportError(CameraException ex) { if (!_isOnline) return; // 标记离线并更新状态为重连中 _isOnline = false; UpdateStatus(VideoSourceStatus.Reconnecting, $"SDK异常: {ex.Message}", ex); } /// /// 标记数据接收(心跳保活 + 双路统计) /// 调用规则: /// 1. 网络层收到流数据时:调用 MarkFrameReceived(dwBufSize),只统计流量。 /// 2. 解码层流控通过后:调用 MarkFrameReceived(0),只统计有效帧率。 /// /// 数据包大小(字节),0 表示这是一帧解码后的图像 protected void MarkFrameReceived(uint dataSize = 0) { long now = Environment.TickCount64; // 1. [心跳保活] 无论网络包还是解码帧,都视为设备“活着” // 使用 Interlocked 保证多线程读写安全 Interlocked.Exchange(ref _lastFrameTick, now); // 2. [分流累加] 根据来源不同,累加不同的计数器 if (dataSize > 0) { // --- 来源:网络层回调 (SafeOnRealDataReceived) --- // 只累加字节数,用于计算带宽 (Mbps) // 绝对不能在这里累加帧数,否则会被网络包的数量误导(导致 FPS 虚高) Interlocked.Add(ref _tempByteCounter, dataSize); } else { // --- 来源:解码层回调 (SafeOnDecodingCallBack) --- // 只累加帧数,用于计算有效 FPS // 只有经过 MakeDecision() 筛选保留下来的帧才走到这里,所以是真实的 "Output FPS" Interlocked.Increment(ref _tempFrameCounter); // 累加生命周期总帧数 Interlocked.Increment(ref _totalFramesReceived); } // 3. [定期结算] 每 1000ms (1秒) 结算一次统计指标 long timeDiff = now - _lastFpsCalcTick; if (timeDiff >= 1000) { // 忽略第一次冷启动的数据(避免除以 0 或时间跨度过大) if (_lastFpsCalcTick > 0) { double duration = timeDiff / 1000.0; // --- A. 结算有效帧率 (FPS) --- // 原子读取并重置计数器,防止漏算 int frames = Interlocked.Exchange(ref _tempFrameCounter, 0); RealFps = Math.Round(frames / duration, 1); // --- B. 结算网络带宽 (Mbps) --- // 公式: (字节数 * 8位) / 1024 / 1024 / 秒数 long bytes = Interlocked.Exchange(ref _tempByteCounter, 0); _currentBitrate = Math.Round((bytes * 8.0) / 1024 / 1024 / duration, 2); } else { // 初始化重置 _tempFrameCounter = 0; _tempByteCounter = 0; } // 更新结算时间锚点 _lastFpsCalcTick = now; } } /// /// 触发帧回调事件(热路径优化) /// /// 帧数据(如 Mat/SmartFrame) protected void RaiseFrameReceived(object frameData) => FrameReceived?.Invoke(frameData); /// /// 后台状态分发循环(单线程消费状态队列) /// /// 取消令牌 private async Task StatusDistributorLoopAsync(CancellationToken token) { try { // 关键修复:使用 CancellationToken.None 等待读取 // 确保取消时仍能消费完队列剩余消息 while (await _statusQueue.Reader.WaitToReadAsync(CancellationToken.None).ConfigureAwait(false)) { // 批量读取队列中的所有消息 while (_statusQueue.Reader.TryRead(out var args)) { // 异常隔离:捕获订阅者回调异常,防止分发器崩溃 try { StatusChanged?.Invoke(this, args); } catch (Exception ex) { Debug.WriteLine($"[UIEventError] 设备 {Id} 状态回调异常: {ex.Message}"); } // 退出条件:取消令牌已触发 且 队列为空 if (token.IsCancellationRequested && _statusQueue.Reader.Count == 0) { return; } } // 双重检查退出条件 if (token.IsCancellationRequested && _statusQueue.Reader.Count == 0) { return; } } } catch (Exception ex) { Debug.WriteLine($"[DistributorFatal] 设备 {Id} 状态分发器崩溃: {ex.Message}"); } } /// /// 更新设备状态并写入分发队列 /// /// 新状态 /// 状态描述 /// 关联异常(可选) protected void UpdateStatus(VideoSourceStatus status, string msg, CameraException? ex = null) { lock (_stateSyncRoot) { // 更新内部状态 _status = status; // 写入状态队列(满时自动丢弃旧数据) _ = _statusQueue.Writer.TryWrite(new StatusChangedEventArgs( status, msg, ex, ex?.RawErrorCode) { NewHandle = ex?.Context.TryGetValue("NativeHandle", out var handle) == true ? (IntPtr)handle : null }); } } #endregion #region --- 10. 审计日志辅助 (Audit Log Helpers) --- /// /// 添加审计日志(线程安全) /// /// 日志内容 public void AddAuditLog(string message) { lock (_auditLogs) { var logEntry = $"[{DateTime.Now:HH:mm:ss.fff}] {message}"; _auditLogs.Add(logEntry); // 滚动清除旧日志 if (_auditLogs.Count > MaxAuditLogCount) { _auditLogs.RemoveAt(0); } } } /// /// 获取审计日志副本 /// /// 日志列表副本 public List GetAuditLogs() { lock (_auditLogs) { return new List(_auditLogs); } } #endregion #region --- 11. 抽象方法 (Abstract Methods) --- /// /// 驱动层启动逻辑(子类必须实现) /// 包含:设备登录、码流订阅、取流线程启动等 /// /// 取消令牌 protected abstract Task OnStartAsync(CancellationToken token); /// /// 驱动层停止逻辑(子类必须实现) /// 包含:码流停止、设备登出、资源释放等 /// protected abstract Task OnStopAsync(); /// /// 驱动层元数据获取逻辑(子类必须实现) /// /// 设备元数据 protected abstract Task OnFetchMetadataAsync(); #endregion #region --- 12. 资源清理 (Resource Disposal) --- /// /// 同步销毁入口(死锁免疫) /// public void Dispose() { // 异步销毁在后台执行,避免阻塞 UI 线程 Task.Run(async () => await DisposeAsync().ConfigureAwait(false)).GetAwaiter().GetResult(); } /// /// 异步销毁资源(优雅关闭) /// /// ValueTask public virtual async ValueTask DisposeAsync() { // 1. 停止业务逻辑 await StopAsync().ConfigureAwait(false); // 2. 优雅关闭状态分发器 _statusQueue.Writer.TryComplete(); // 标记队列不再接受新消息 _distributorCts?.Cancel(); // 触发分发器取消 // 3. 等待分发器处理完剩余消息(最多等待 500ms) if (_distributorTask != null) { await Task.WhenAny(_distributorTask, Task.Delay(500)).ConfigureAwait(false); } // 4. 切断事件引用,防止内存泄漏 FrameReceived = null; StatusChanged = null; // 5. 释放基础资源 _lifecycleLock.Dispose(); _distributorCts?.Dispose(); // 6. 抑制垃圾回收器的终结器 GC.SuppressFinalize(this); } #endregion #region --- 13. 内部字段与补全属性 --- /// 跟踪上一个未完成的生命周期任务 private Task _pendingLifecycleTask = Task.CompletedTask; /// 帧控制器(用于帧分发策略管理) public FrameController Controller { get; protected set; } #endregion // 自动从 SmartFrame 中提取 public int Width { get; protected set; } public int Height { get; protected set; } public void ClearAuditLogs() { _auditLogs.Clear(); AddAuditLog("用户清空了审计日志"); } }