using System;
|
using System.Diagnostics;
|
using System.Threading;
|
using System.Threading.Tasks;
|
using Aging.Infrastructure.Broker;
|
using Aging.ResourceContracts;
|
using log4net;
|
|
namespace Aging.Infrastructure.Drivers
|
{
|
/// <summary>
|
/// Broker 传输后端：命令经命名管道 RPC 委托 DeviceBroker 宿主执行（物理会话由宿主唯一持有）。
|
/// 三重关卡（白名单/configHash/租约栅栏）与 SessionGate 全在宿主侧，本类只做提交与态映射。
|
/// 接收帧由宿主随响应以 "|PUSHED:n|EV:RECV…" 后缀捎带返回，解析后经 FramePushed 上抛（§5.4 捎带模型）。
|
/// </summary>
|
internal sealed class BrokerRpcTransport : ICanTransport
|
{
|
private static readonly ILog Log = LogManager.GetLogger(typeof(BrokerRpcTransport));
|
|
private readonly BrokerClientChannel _channel;
|
private readonly string _deviceId;
|
private readonly string _configHash;
|
private bool _disposed;
|
|
public bool IsConnected => _channel.IsConnected;
|
public event Action<string> FramePushed;
|
public event Action TransportBroken;
|
|
public BrokerRpcTransport(string deviceId, string pipeName, string configHash)
|
{
|
_deviceId = deviceId ?? throw new ArgumentNullException(nameof(deviceId));
|
_configHash = configHash ?? "";
|
string instanceId = $"{Environment.MachineName}/{Process.GetCurrentProcess().Id}";
|
_channel = new BrokerClientChannel(pipeName, instanceId);
|
}
|
|
// busName/nodeIndex 对 Broker 无意义（物理会话在宿主按设备端点建立），CONNECT 即注册
|
public Task ConnectAsync(string busName, int nodeIndex, CancellationToken ct)
|
=> SubmitExpectOkAsync("CONNECT", "", ct);
|
|
public Task DisconnectAsync()
|
=> SubmitExpectOkAsync("DISCONNECT", "", CancellationToken.None);
|
|
public async Task<string> TransceiveAsync(string line, CancellationToken ct)
|
{
|
if (_disposed) throw new ObjectDisposedException(nameof(BrokerRpcTransport));
|
|
// line 形如 "SEND:…" 或 "QUERY:…"，映射回宿主白名单 opCode + payload：
|
// SEND → 宿主的 device.Driver.SendCommandAsync 需完整 "SEND:frameId:data"，payload 保留前缀
|
// QUERY → 宿主的 device.Driver.QueryAsync 需去 QUERY: 前缀的裸查询
|
string opCode;
|
string payload;
|
if (line.StartsWith("SEND:", StringComparison.Ordinal))
|
{
|
opCode = "SEND";
|
payload = line;
|
}
|
else if (line.StartsWith("QUERY:", StringComparison.Ordinal))
|
{
|
opCode = "QUERY";
|
payload = line.Substring("QUERY:".Length);
|
}
|
else
|
{
|
opCode = "QUERY";
|
payload = line;
|
}
|
|
var result = await SubmitAsync(opCode, payload, ct).ConfigureAwait(false);
|
DrainPiggyback(result.Readback, out string clean);
|
return clean;
|
}
|
|
private async Task SubmitExpectOkAsync(string opCode, string payload, CancellationToken ct)
|
{
|
var r = await SubmitAsync(opCode, payload, ct).ConfigureAwait(false);
|
// CONNECT/DISCONNECT 无有效读回帧，忽略捎带
|
_ = r.Readback;
|
}
|
|
private async Task<BrokerIpc.ResultBody> SubmitAsync(string opCode, string payload, CancellationToken ct)
|
{
|
try
|
{
|
await _channel.EnsureConnectedAsync(ct).ConfigureAwait(false);
|
var body = new BrokerIpc.SubmitBody
|
{
|
DeviceId = _deviceId,
|
OperationCode = opCode,
|
Payload = payload ?? "",
|
ResourceKey = _deviceId,
|
ClientInstanceId = $"{Environment.MachineName}/{Process.GetCurrentProcess().Id}"
|
};
|
var r = await _channel.SubmitAsync(body, _configHash, ct).ConfigureAwait(false);
|
ThrowForState(opCode, r);
|
return r;
|
}
|
catch (Exception ex) when (!(ex is ObjectDisposedException))
|
{
|
Log.Error($"[BrokerRpcCAN] {_deviceId} {opCode} 失败: {ex.Message}", ex);
|
throw;
|
}
|
}
|
|
/// <summary>从 readback 尾部剥离 |PUSHED:n|EV:RECV… 段并触发 FramePushed。</summary>
|
private void DrainPiggyback(string readback, out string clean)
|
{
|
clean = readback ?? "";
|
if (string.IsNullOrEmpty(clean)) return;
|
|
int idx = clean.IndexOf("|PUSHED:", StringComparison.Ordinal);
|
if (idx < 0) return;
|
|
string pushSection = clean.Substring(idx + "|PUSHED:".Length);
|
clean = clean.Substring(0, idx);
|
|
// pushSection 形如 "n|EV:RECV:…|EV:RECV:…"
|
var segs = pushSection.Split('|');
|
for (int i = 1; i < segs.Length; i++)
|
{
|
if (segs[i].StartsWith("EV:", StringComparison.Ordinal))
|
FramePushed?.Invoke(segs[i]);
|
}
|
}
|
|
private void ThrowForState(string opCode, BrokerIpc.ResultBody r)
|
{
|
switch (r.State)
|
{
|
case "Succeeded":
|
return;
|
case "RejectedStaleLease":
|
throw new InvalidOperationException($"LEASE_STALE: {_deviceId} 租约栅栏校验失败: {r.ErrorCode} {r.Readback}");
|
case "RejectedConfigMismatch":
|
throw new InvalidOperationException($"CONFIG_MISMATCH: Broker 生效配置与本地快照不一致(INV-04): {r.Readback}");
|
case "RejectedQueueFull":
|
throw new InvalidOperationException($"QUEUE_FULL: {r.Readback}");
|
case "Unknown":
|
throw new TimeoutException($"UNKNOWN: {_deviceId} 动作 {opCode} 结果未知，禁止重试假定未执行(INV-06): {r.Readback}");
|
default:
|
throw new InvalidOperationException($"Broker 执行失败[{r.ErrorCode}]: {r.Readback}");
|
}
|
}
|
|
public void Dispose()
|
{
|
if (_disposed) return;
|
_disposed = true;
|
_channel.Dispose();
|
}
|
}
|
}
|
