root/Ayay
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

NetMQ 协议，支持摄像头增、删、改

Browse Source
This commit is contained in:
wilson
2026-01-12 18:27:58 +08:00
parent 031d4f3416
commit 3f8e42e560
20 changed files with 604 additions and 332 deletions
Download Patch File Download Diff File Expand all files Collapse all files

125
SHH.CameraService/Core/CmdClients/CommandClientWorker.cs
View File

@@ -1,10 +1,11 @@
using MessagePack;
using System.Text;
using MessagePack;
using Microsoft.Extensions.Hosting;
using NetMQ;
using NetMQ.Monitoring; // ★ 1. 必须引用 Monitoring 命名空间
using NetMQ.Sockets;
using SHH.CameraSdk;
using SHH.Contracts;
using System.Text;
namespace SHH.CameraService;
@@ -16,6 +17,10 @@ public class CommandClientWorker : BackgroundService
// 管理多个 Socket
private readonly List<DealerSocket> _sockets = new();
// ★ 2. 新增：保存 Monitor 列表，防止被 GC 回收
private readonly List<NetMQMonitor> _monitors = new();
private NetMQPoller? _poller;
public CommandClientWorker(
@@ -34,71 +39,53 @@ public class CommandClientWorker : BackgroundService
if (!_config.ShouldConnect || _config.CommandEndpoints.Count == 0) return;
// 1. 建立连接 (但不立即启动 Poller)
_poller = new NetMQPoller();
// -------------------------------------------------------------
// 核心修改区：建立连接并挂载监控器
// -------------------------------------------------------------
foreach (var ep in _config.CommandEndpoints)
{
try
{
var socket = new DealerSocket();
// 建议加上 Socket 索引或 UUID 以防服务端认为 Identity 冲突
// 或者保持原样，取决于服务端逻辑。通常同一个 AppId 连不同 Server 是没问题的。
socket.Options.Identity = Encoding.UTF8.GetBytes(_config.AppId);
socket.Connect(ep.Uri);
var monitorUrl = $"inproc://monitor_{Guid.NewGuid():N}";
var monitor = new NetMQMonitor(socket, monitorUrl, SocketEvents.Connected);
monitor.Connected += async (s, args) =>
{
Console.WriteLine($"[指令] 网络连接建立: {ep.Uri} -> 正在补发注册包...");
await SendRegisterAsync(socket);
};
// ★★★ 修正点：使用 AttachToPoller 代替 Add ★★★
// 错误写法: _poller.Add(monitor);
monitor.AttachToPoller(_poller);
// 依然需要保存引用，防止被 GC 回收
_monitors.Add(monitor);
socket.Connect(ep.Uri);
socket.ReceiveReady += OnSocketReceiveReady;
_sockets.Add(socket);
_poller.Add(socket);
Console.WriteLine($"[指令] 建立通道: {ep.Uri}");
Console.WriteLine($"[指令] 通道初始化完成: {ep.Uri} (带自动重连监控)");
}
catch (Exception ex)
{
Console.WriteLine($"[指令] 连接初始化异常: {ex.Message}");
}
catch (Exception ex) { Console.WriteLine($"[指令] 连接异常: {ex.Message}"); }
}
if (_sockets.Count == 0) return;
// =================================================================
// 2. 发送注册包 (在 Poller 启动前发送，绝对线程安全)
// 6. 绑定 ACK 逻辑 (保持不变)
// =================================================================
var registerPayload = new RegisterPayload
{
Protocol = ProtocolHeaders.ServerRegister,
InstanceId = _config.AppId,
ProcessId = Environment.ProcessId,
Version = "1.0.0",
ServerIp = "127.0.0.1",
WebApiPort = _config.BasePort,
StartTime = DateTime.Now
};
try
{
byte[] regData = MessagePackSerializer.Serialize(registerPayload);
var ctx = await _pipeline.ExecuteSendAsync(ProtocolHeaders.ServerRegister, regData);
if (ctx != null)
{
foreach (var socket in _sockets)
{
// 此时 Poller 还没跑，主线程发送是安全的
socket.SendMoreFrame(ctx.Protocol).SendFrame(ctx.Data);
}
Console.WriteLine($"[指令] 注册包已广播至 {_sockets.Count} 个目标");
}
}
catch (Exception ex)
{
Console.WriteLine($"[指令] 注册失败: {ex.Message}");
}
// =================================================================
// 3. 绑定 ACK 逻辑
// =================================================================
// 关键修正：直接使用 async void，不要包裹在 Task.Run 中！
// 因为 OnResponseReady 是由 Dispatcher 触发的，而 Dispatcher 是由 Poller 线程触发的。
// 所以这里就在 Poller 线程内，可以直接操作 Socket。
_dispatcher.OnResponseReady += async (result) =>
{
try
@@ -122,8 +109,11 @@ public class CommandClientWorker : BackgroundService
};
// =================================================================
// 4. 启动 Poller (开始监听接收)
// 7. 启动 Poller
// =================================================================
// 注意：我们不需要手动发第一次注册包了，
// 因为 Poller 启动后，底层 TCP 会建立连接，从而触发 monitor.Connected 事件，
// 事件里会自动发送注册包。这就是“自动档”的好处。
_poller.RunAsync();
// 阻塞直到取消
@@ -135,12 +125,49 @@ public class CommandClientWorker : BackgroundService
// 清理
_poller.Stop();
_poller.Dispose();
foreach (var m in _monitors) m.Dispose(); // 释放监控器
foreach (var s in _sockets) s.Dispose();
}
// =================================================================
// ★ 8. 抽离出的注册包发送逻辑 (供 Monitor 调用)
// =================================================================
private async Task SendRegisterAsync(DealerSocket targetSocket)
{
try
{
var registerPayload = new RegisterPayload
{
Protocol = ProtocolHeaders.ServerRegister,
InstanceId = _config.AppId,
ProcessId = Environment.ProcessId,
Version = "1.0.0",
ServerIp = "127.0.0.1", // 建议优化：获取本机真实IP
WebApiPort = _config.BasePort,
StartTime = DateTime.Now
};
byte[] regData = MessagePackSerializer.Serialize(registerPayload);
// 执行拦截器
var ctx = await _pipeline.ExecuteSendAsync(ProtocolHeaders.ServerRegister, regData);
if (ctx != null)
{
// 直接向触发事件的那个 Socket 发送
// DealerSocket 允许在连接未完全就绪时 Send，它会缓存直到网络通畅
targetSocket.SendMoreFrame(ctx.Protocol).SendFrame(ctx.Data);
// Console.WriteLine($"[指令] 身份注册包已推入队列: {targetSocket.Options.Identity}");
}
}
catch (Exception ex)
{
Console.WriteLine($"[指令] 注册包发送失败: {ex.Message}");
}
}
private async void OnSocketReceiveReady(object? sender, NetMQSocketEventArgs e)
{
// 这里的代码运行在 Poller 线程
NetMQMessage incomingMsg = new NetMQMessage();
if (e.Socket.TryReceiveMultipartMessage(ref incomingMsg))
{
@@ -154,8 +181,6 @@ public class CommandClientWorker : BackgroundService
var ctx = await _pipeline.ExecuteReceiveAsync(rawProtocol, rawData);
if (ctx != null)
{
// DispatchAsync 会同步触发 OnResponseReady，
// 从而在同一个线程内完成 ACK 发送，线程安全且高效。
await _dispatcher.DispatchAsync(ctx.Protocol, ctx.Data);
}
}

84
SHH.CameraService/Core/CmdClients/RemoveCameraHandler.cs Normal file
View File

@@ -0,0 +1,84 @@
using Newtonsoft.Json.Linq;
using SHH.CameraSdk;
using SHH.Contracts;
namespace SHH.CameraService
{
/// <summary>
/// 移除设备指令处理器
/// </summary>
public class RemoveCameraHandler : ICommandHandler
{
private readonly CameraManager _cameraManager;
/// <summary>
/// 指令名称
/// </summary>
public string ActionName => ProtocolHeaders.Remove_Camera;
/// <summary>
/// 构造函数
/// </summary>
/// <param name="cameraManager"></param>
public RemoveCameraHandler(CameraManager cameraManager)
{
_cameraManager = cameraManager;
}
/// <summary>
/// 处理指令
/// </summary>
/// <param name="payload"></param>
public async Task ExecuteAsync(JToken payload)
{
long deviceId = 0;
try
{
// 1. 增强型 ID 解析
if (payload.Type == JTokenType.Object)
{
// 兼容大小写不敏感的解析
var idToken = payload["Id"] ?? payload["id"];
if (idToken != null) deviceId = idToken.Value<long>();
}
else if (payload.Type == JTokenType.Integer || payload.Type == JTokenType.String)
{
// 兼容字符串形式的 ID
long.TryParse(payload.ToString(), out deviceId);
}
if (deviceId <= 0)
{
Console.WriteLine($"[{ActionName}] 收到无效指令: ID解析失败 ({payload})");
return;
}
// 2. 预检查
var device = _cameraManager.GetDevice(deviceId);
if (device == null)
{
Console.WriteLine($"[{ActionName}] 设备 {deviceId} 已经不在管理池中，无需操作。");
return;
}
// 3. 安全移除
// 这里建议增加审计日志，记录谁触发了删除（如果协议里有用户信息的话）
device.AddAuditLog("收到远程指令：彻底移除设备");
Console.WriteLine($"[{ActionName}] 正在安全移除设备: {deviceId} ({device.Config.Name})");
// CameraManager 内部会：StopAsync -> DisposeAsync -> TryRemove -> SaveChanges
await _cameraManager.RemoveDeviceAsync(deviceId);
Console.WriteLine($"[{ActionName}] 设备 {deviceId} 已彻底清理并从持久化库中移除。");
// 4. (可选) 此处可以调用 CommandDispatcher 发送 Success ACK
}
catch (Exception ex)
{
// 捕获异常，防止影响全局 Socket 轮询
Console.WriteLine($"[{ActionName}] 移除设备 {deviceId} 过程中发生致命错误: {ex.Message}");
}
}
}
}

212
SHH.CameraService/Core/CmdClients/SyncCameraHandler.cs
View File

@@ -1,103 +1,177 @@
using Newtonsoft.Json.Linq;
using SHH.CameraSdk; // 引用包含 FrameController 和 FrameRequirement 的命名空间
using SHH.CameraSdk;
using SHH.Contracts;
namespace SHH.CameraService;
/// <summary>
/// 同步设备配置处理器
/// </summary>
public class SyncCameraHandler : ICommandHandler
{
private readonly CameraManager _cameraManager;
public string ActionName => ProtocolHeaders.SyncCamera;
/// <summary>
/// 命令名称
/// </summary>
public string ActionName => ProtocolHeaders.Sync_Camera;
/// <summary>
/// 构造函数
/// </summary>
/// <param name="cameraManager"></param>
public SyncCameraHandler(CameraManager cameraManager)
{
_cameraManager = cameraManager;
}
/// <summary>
/// 执行处理
/// </summary>
/// <param name="payload"></param>
/// <returns></returns>
public async Task ExecuteAsync(JToken payload)
{
// 1. 解析配置
// 1. 反序列化配置 DTO
var dto = payload.ToObject<CameraConfigDto>();
if (dto == null) return;
// 2. 添加设备到管理器 (这一步是必须的，不然没有 Device 就没有 Controller)
var videoConfig = new VideoSourceConfig
{
Id = dto.Id,
Name = dto.Name,
IpAddress = dto.IpAddress,
Port = dto.Port,
Username = dto.Username,
Password = dto.Password,
ChannelIndex = dto.ChannelIndex,
StreamType = dto.StreamType,
Brand = (DeviceBrand)dto.Brand,
RenderHandle = (IntPtr)dto.RenderHandle,
MainboardIp = dto.MainboardIp,
MainboardPort = dto.MainboardPort,
// 必须给个默认值，防止空引用
VendorArguments = new Dictionary<string, string>(),
};
// 如果设备不存在才添加，如果已存在，后续逻辑会直接获取
if (_cameraManager.GetDevice(videoConfig.Id) == null)
{
_cameraManager.AddDevice(videoConfig);
}
// 3. 核心：直接获取设备实例
// 2. 尝试获取现有设备
var device = _cameraManager.GetDevice(dto.Id);
if (device == null)
{
Console.WriteLine($"[SyncError] 设备 {dto.Id} 创建失败，无法执行自动订阅。");
return;
}
// 4. 拿到你的“宝贝”控制器 (FrameController)
var controller = device.Controller;
if (controller == null)
if (device != null)
{
Console.WriteLine($"[SyncError] 设备 {dto.Id} 不支持流控调度 (Controller is null)。");
return;
}
// =========================================================
// 场景 A: 设备已存在 -> 执行智能更新 (Smart Update)
// =========================================================
Console.WriteLine($"[Sync] 更新设备配置: {dto.Id} ({dto.Name})");
// 5. 暴力注册订阅需求 (Loop AutoSubscriptions)
if (dto.AutoSubscriptions != null && dto.AutoSubscriptions.Count > 0)
{
foreach (var subItem in dto.AutoSubscriptions)
// 将全量配置映射为部分更新 DTO
var updateDto = new DeviceUpdateDto
{
// 生成 AppId (照抄你给的逻辑)
string finalAppId = string.IsNullOrWhiteSpace(subItem.AppId)
? $"SUB_{Guid.NewGuid().ToString("N").Substring(0, 8).ToUpper()}"
: subItem.AppId;
// --- 冷更新参数 (变更会触发重启) ---
IpAddress = dto.IpAddress,
Port = dto.Port,
Username = dto.Username,
Password = dto.Password,
ChannelIndex = dto.ChannelIndex,
Brand = dto.Brand,
RtspPath = dto.RtspPath,
RenderHandle = dto.RenderHandle, // long 类型直接赋值
Console.WriteLine($"[自动化] 正在注册流控: {finalAppId}, 目标: {subItem.TargetFps} FPS");
// --- 热更新参数 (变更立即生效) ---
Name = dto.Name,
Location = dto.Location,
StreamType = dto.StreamType,
// 构造 FrameRequirement 对象 (完全匹配你 FrameController 的入参)
// 这里的属性赋值对应你代码里 req.Type, req.SavePath 等逻辑
var requirement = new FrameRequirement
MainboardIp = dto.MainboardIp,
MainboardPort = dto.MainboardPort,
// --- 图像处理参数 (热更新) ---
AllowCompress = dto.AllowCompress,
AllowExpand = dto.AllowExpand,
TargetResolution = dto.TargetResolution,
EnhanceImage = dto.EnhanceImage,
UseGrayscale = dto.UseGrayscale
};
// 调用 Manager 的核心更新逻辑 (它会自动判断是 Stop->Start 还是直接应用)
await _cameraManager.UpdateDeviceConfigAsync(dto.Id, updateDto);
}
else
{
// =========================================================
// 场景 B: 设备不存在 -> 执行新增 (Add New)
// =========================================================
Console.WriteLine($"[Sync] 新增设备: {dto.Id} ({dto.Name})");
// 构造全新的设备配置
var newConfig = new VideoSourceConfig
{
Id = dto.Id,
Name = dto.Name,
Brand = (DeviceBrand)dto.Brand, // int -> Enum 强转
IpAddress = dto.IpAddress,
Port = dto.Port,
Username = dto.Username,
Password = dto.Password,
ChannelIndex = dto.ChannelIndex,
StreamType = dto.StreamType,
RtspPath = dto.RtspPath,
MainboardIp = dto.MainboardIp,
MainboardPort = dto.MainboardPort,
RenderHandle = (IntPtr)dto.RenderHandle, // long -> IntPtr 转换
ConnectionTimeoutMs = 5000 // 默认超时
};
// 添加到管理器池
_cameraManager.AddDevice(newConfig);
// 重新获取引用以进行后续操作
device = _cameraManager.GetDevice(dto.Id);
}
// ★★★ 核心修复：统一处理“运行意图” ★★★
if (device != null)
{
// 将 DTO 的立即执行标志直接同步给设备的运行意图
device.IsRunning = dto.ImmediateExecution;
if (dto.ImmediateExecution)
{
// 情况 1: 收到“启动”指令
if (!device.IsOnline) // 只有没在线时才点火
{
AppId = finalAppId,
TargetFps = subItem.TargetFps, // 8帧 或 1帧
Type = (SubscriptionType)subItem.Type, // 业务类型 (LocalWindow, NetworkTrans...)
Memo = subItem.Memo ?? "Auto Sync",
// 其它字段给默认空值，防止 Controller 内部逻辑报错
Handle = "",
SavePath = ""
};
// ★★★ 见证奇迹的时刻：直接调用 Register ★★★
controller.Register(requirement);
Console.WriteLine($"[Sync] 指令：立即启动设备 {dto.Id}");
_ = device.StartAsync();
}
}
else
{
// 情况 2: 收到“停止”指令 (即 ImmediateExecution = false)
if (device.IsOnline) // 只有在线时才熄火
{
Console.WriteLine($"[Sync] 指令：立即停止设备 {dto.Id}");
_ = device.StopAsync();
}
}
}
//// 6. 启动设备
//// 你的积分算法会在 device 内部的推流循环中被 MakeDecision 调用
if (dto.ImmediateExecution)
await device.StartAsync();
// =========================================================
// 3. 处理自动订阅策略 (Auto Subscriptions)
// =========================================================
// 无论新增还是更新，都确保订阅策略是最新的
if (device != null && dto.AutoSubscriptions != null)
{
var controller = device.Controller;
if (controller != null)
{
foreach (var sub in dto.AutoSubscriptions)
{
// 如果没有 AppId，生成一个临时的（通常 Dashboard 会下发固定的 AppId）
string appId = string.IsNullOrWhiteSpace(sub.AppId)
? $"AUTO_{Guid.NewGuid().ToString("N")[..8]}"
: sub.AppId;
Console.WriteLine($"[SyncSuccess] 设备 {dto.Id} 同步完成，策略已下发。");
// 构造流控需求
var req = new FrameRequirement
{
AppId = appId,
TargetFps = sub.TargetFps,
Type = (SubscriptionType)sub.Type, // int -> Enum
Memo = sub.Memo ?? "Sync Auto",
// 自动订阅通常不包含具体的 Handle 或 SavePath，除非协议里带了
// 如果需要支持网络转发，这里可以扩展映射 sub.TargetIp 等
Handle = "",
SavePath = ""
};
// 注册到帧控制器
controller.Register(req);
}
}
}
}
}