OpenRobot.Framework.Communication 1.0.0

OpenRobot.Framework.Communication

MQTT 连接与 IPC 跨进程通讯通用库（net8.0，单包内含 Mqtt / Ipc 两个独立命名空间）。

• MQTT：连接控制 + 自动重连 + 重连后重订阅 + 按序补发；订阅以 (Topic, Handler) 批量注册，收包经 Channel 消费队列派发；QoS / Retain / MQTT5 ResponseTopic 与 CorrelationData 可配。
• IPC：一个服务端 + 多个客户端。服务端按握手 ClientId 维护会话表，定向发送 / 广播 / 单会话断开互不影响；Client 与 Session 接口同构，双向「单向发送（不等待返回）」与「请求-响应（带超时）」都可用；传输可在命名管道 / TCP 之间切换。
• 协议栈类型（MQTTnet）不出现在公共 API：换掉 MQTTnet 不影响调用方。
• 日志全部经 ICommunicationLogger，宿主可实现后接管输出（也支持直接用 Microsoft.Extensions.Logging）。
• 无副作用：不写环境变量、不落盘、不生成持久状态；凭据只读，全部由宿主提供。

安装

<PackageReference Include="OpenRobot.Framework.Communication" Version="1.0.0" />

包由本仓库产出：dotnet pack common/OpenRobot.Framework.Communication/OpenRobot.Framework.Communication.csproj -c Release， 产物为 OpenRobot.Framework.Communication.1.0.0.nupkg（lib/net8.0 单个程序集）与配套 .snupkg。

依赖：

• MQTTnet 4.3.7.1207（仅内部使用）
• Microsoft.Extensions.Logging.Abstractions 8.0.2
• Microsoft.Extensions.DependencyInjection.Abstractions 8.0.2

快速开始

MQTT

using OpenRobot.Framework.Communication;
using OpenRobot.Framework.Communication.Mqtt;

var mqtt = new MqttClient(new MqttOptions
{
    Name = "primary",
    Broker = "127.0.0.1",
    Port = 1883,
    Logger = new ConsoleCommunicationLogger(),   // 或实现 ICommunicationLogger
});

mqtt.AddSubscriptions(
    // 处理完要回执：返回值非 null 即回执（带 CancellationToken 的重载用 (m, ct) => ...）
    new MqttSubscription("ota/down/#", HandleOtaDownAsync),
    new MqttSubscription("config/changed", msg => Console.WriteLine(msg.Payload)));

await mqtt.StartAsync();     // 非阻塞：见下方「StartAsync 不等首连」
await Task.Delay(500);
await mqtt.PublishAsync("ota/up", new { deviceId = "d1", state = "ok" });

static Task<string?> HandleOtaDownAsync(MqttMessage msg, CancellationToken ct)
    => Task.FromResult<string?>("{\"accepted\":true}");

IPC 服务端（一个端点，多个客户端）

using OpenRobot.Framework.Communication.Ipc;

await using var service = new IpcService(new IpcOptions { Name = "hub", PipePrefix = "EdgeControl" });

service.On("telemetry.report", inbound =>
{
    // inbound.SourceId 是发来这条消息的客户端 ClientId
    Console.WriteLine($"{inbound.SourceId}: {inbound.Message.Payload}");
    return Task.CompletedTask;
});
// 需要强类型载荷时显式指定泛型参数（否则 (payload, ctx) 双参 lambda 无法推断 T）
service.On<TelemetryReport>("telemetry.report", report => HandleReportAsync(report));

service.OnRequest("api.proxy.request", async ctx =>
{
    if (!await IsAuthorizedAsync(ctx.SourceId)) { await ctx.ReplyErrorAsync("unauthorized"); return; }
    await ctx.ReplyAsync(new { ok = true });
});

service.ClientConnected += (_, e) => Console.WriteLine($"上线: {e.Session.ClientId} @ {e.Session.RemoteEndpoint}");

await service.StartAsync();                       // 返回时已可连接
await service.SendToAsync("upgrader", "ota.dispatch", new { url = "https://..." });
await service.BroadcastAsync("config.changed", new { revision = 42 });
var reply = await service.RequestAsync("upgrader", "api.proxy.request", new { path = "/x" },
    TimeSpan.FromSeconds(30));

IPC 客户端（独立进程）

await using var client = new IpcClient(new IpcOptions
{
    Name = "upgrader",
    ClientId = "upgrader",        // 服务端按它区分会话
    PipePrefix = "EdgeControl",
    DefaultRequestTimeout = TimeSpan.FromSeconds(60),
});

client.On<OtaDispatch>("ota.dispatch", async task => await RunAsync(task));   // 不关心来源
await client.StartAsync();

// 客户端也能反向往服务端发请求（双向对称）
var cfg = await client.RequestAsync("config.get", new { key = "revision" });

API 速查

IMqttClient

MqttSubscription 四个构造重载：Func<MqttMessage, CancellationToken, Task<string?>> 与 Func<MqttMessage, Task<string?>>（返回值非 null 且消息带 ResponseTopic 时自动回执）、Func<MqttMessage, Task>、Action<MqttMessage>（后两者不回复）。

带 Task<string?> 的两个重载必须都在：只有三参版时，一参 lambda m => Task.FromResult<string?>(json) 会因元数对不上而静默绑定到「不回执」的 Func<MqttMessage, Task>（Task<string?> 可隐式转 Task），回执凭空消失且不报错。

IIpcService（服务端）

IIpcClient / IIpcSession（共用 IIpcEndpoint）

处理器内可用 IpcInboundMessage.SourceId 判断来源；请求处理器用 IpcRequestContext.ReplyAsync / ReplyErrorAsync 回传（ReplyErrorAsync 回 {"success":false,"error":...}，请求方快速失败而非等超时）。

__ipc. 前缀由库独占（握手帧 __ipc.hello），业务消息类型不得占用。

DI 集成

services.AddOpenRobotCommunication(o =>
{
    o.Logger = myLogger;                                   // 全局日志（可选）
    o.AddMqtt("primary", m => { m.Broker = "broker.local"; m.LogTag = "Mqtt.primary"; });
    o.AddIpcService("hub", i => { i.Name = "hub"; i.PipePrefix = "EdgeControl"; i.MaxClients = 32; });
    o.AddIpcClient("upgrader", i => { i.Name = "upgrader"; i.ClientId = "upgrader"; });
});

var client = sp.GetRequiredKeyedService<IMqttClient>("primary");
await client.StartAsync();

• 每个 MQTT 连接 = keyed singleton IMqttClient；每个 IPC 端点 = keyed singleton IIpcClient / IIpcService。
• 不自动连接 / 监听：注册只构造实例，由宿主在适当时机调用 StartAsync。
• 日志优先级：MqttOptions/IpcOptions.Logger > CommunicationOptions.Logger > 容器的 ILoggerFactory > 静默。
• 容器必须异步释放：三个服务都是 IAsyncDisposable 且不实现 IDisposable，同步 ServiceProvider.Dispose() 会抛 InvalidOperationException（"type only implements IAsyncDisposable. Use DisposeAsync to dispose the container."）。用 await using var sp = services.BuildServiceProvider();。

配置项

MqttOptions

IpcOptions（Service 与 Client 共用）

写失败一律拆连接、不在原连接上重试：写失败可能已在线上留下半截帧，重试会把新帧追加进半截帧导致对端静默解析错乱。在途消息保留，由下一条连接优先补发。

平台说明

• Windows 命名管道：端点名为 {PipePrefix}.{Name}（如 EdgeControl.hub），保留原大小写；管道名不区分大小写。同一服务端的管道实例数受系统上限 254 约束（库会为「在途 Accept」与「待拒绝连接」各留 1 个实例，故会话数实际封顶 252）。需要更多并发会话请改用 Transport = Tcp。
• Linux / macOS：System.IO.Pipes 自动映射为 Unix Domain Socket，端点名为 /tmp/{prefix}.{name}.sock（转小写——UDS 区分大小写，且路径总长受系统上限约 108 字节约束）。PipePrefix 与 Name 应保持简短，过长会导致连接失败。
• TCP：跨机器可用，但本期为明文传输，无 TLS。仅建议在受信内网使用；需要跨不可信网络时应在链路上自行加隧道 / 加密，或等待后续 TLS 扩展点。
• 传输可测性：IIpcTransport 是接缝，InMemoryTransport 用于单进程内的确定性测试（TransportOverride 注入）。

已知限制

1. UnsubscribeAsync 不发 UNSUBSCRIBE：内部 broker 接缝（IMqttBrokerConnection）只提供 Connect / Disconnect / Publish / Subscribe，没有 Unsubscribe。因此"注销订阅"只做「从注册表摘除 + 移出重订阅清单」——broker 可能仍在投递，但注册表中已无匹配项，派发器只记 Warn 而不调用任何处理器。需要真实退订需先扩接缝。
2. MqttClient.StartAsync 不等首次连接：它只启动后台重连循环，首连失败不向上抛。await StartAsync() 返回后 IsConnected 仍可能为 false。需要「连上再继续」请用 ConnectAsync，或订阅 ConnectionStateChanged 等 Connected。宿主的启动就绪判断必须基于 State / ConnectionStateChanged。
3. 容器需异步释放：见「DI 集成」一节。
4. MQTT 消息顺序仅在 DispatcherConsumerCount = 1（默认）时保证；提高并发数即放弃该保证。
5. IPC 单向发送不保证送达：SendAsync 只入队即返回，队列满时丢最旧（记 Warn）；需要确认送达请用 RequestAsync。
6. IPC 服务端不缓存离线消息：SendToAsync 对离线客户端返回 false；BroadcastAsync 跳过离线会话。
7. IPC 握手不可省略：客户端连接后第一帧必须是 __ipc.hello，首帧非 hello / ClientId 为空 / 帧非法 / 超 HandshakeTimeoutMs 的连接一律被关闭且不建会话。
8. 凭据只读：库不生成、不持久化、不写环境变量。ClientId 三级回退中最后一级是进程内随机 Guid，多实例场景应显式配置 ClientId。
9. MqttClient 公开构造只有 (MqttOptions, IMqttCredentialProvider?)：协议栈工厂与 logger 的注入构造为 internal（供同程序集与测试使用），宿主无法替换协议栈实现。

版本历史

1.0.0（首次发布）

MQTT

• 连接控制：StartAsync（非阻塞 + 自动重连）/ ConnectAsync / StopAsync，显式状态机与 ConnectionStateChanged 事件
• 断线自动重连，重连后重订阅全部 filter（幂等，不重复 SUBSCRIBE）+ 按序补发（上限 / 重试次数可配）
• 订阅以 (Topic, Handler) 批量注册（TopicFilter 去重），收包经 Channel 消费队列派发，命中全部匹配订阅（按 filter 明确度降序）
• 处理器返回值非 null 且消息带 ResponseTopic 时自动回执（带 CorrelationData）；抛异常自动回错误载荷
• IMqttMessageInterceptor 收包前置钩子（存档 / ACK 等）
• QoS / Retain / ResponseTopic / CorrelationData 透传；IMqttCredentialProvider 只读凭据接入

IPC

• 一个服务端 + 多个客户端：按握手 ClientId 建会话表，定向发送 / 广播 / 请求-响应 / 单会话断开互不影响；重复 ClientId 以新连接为准
• IIpcClient 与 IIpcSession 同构（共用 IIpcEndpoint），双向通讯天然对称
• 单向发送走 Channel 待发队列（不等待写出）；请求-响应带超时（超时返回 null，不抛）
• 客户端断线自动重连，每次重连重发握手帧
• 传输可切命名管道（Windows）/ UDS（Linux、macOS）/ TCP（明文），IIpcTransport 为可替换接缝
• 处理器支持服务级（对所有会话生效）/ 会话级（优先），4 种注册重载（含强类型载荷）

通用

• ICommunicationLogger 日志抽象（含 ConsoleCommunicationLogger / MicrosoftCommunicationLogger / NullCommunicationLogger）
• AddOpenRobotCommunication() DI 扩展（keyed singleton，不自动连接）
• 公共 API 不含 MQTTnet / 源项目类型