using System;
|
using System.Buffers.Binary;
|
using System.Collections.Concurrent;
|
using System.IO;
|
using System.IO.Pipes;
|
using System.Text;
|
using System.Threading;
|
using System.Threading.Tasks;
|
using Aging.ResourceContracts;
|
using log4net;
|
|
namespace Aging.Infrastructure.Broker
|
{
|
/// <summary>
|
/// DeviceBroker 命名管道客户端通道（V3.0 §13 同机基线）。
|
/// 修复现有 PipeClient 的三大问题：响应关联可达、超时生效、取消可传播。
|
/// 帧格式：4 字节大端长度前缀 + UTF8 JSON 封套；请求-响应经 RequestId/CorrelationId 配对。
|
/// 线程模型：单后台读循环 + pending 字典；写入用信号量串行化。线程安全，可被多租约共享。
|
/// </summary>
|
public sealed class BrokerClientChannel : IDisposable
|
{
|
private static readonly ILog Log = LogManager.GetLogger(typeof(BrokerClientChannel));
|
|
private readonly string _pipeName;
|
private readonly string _clientInstanceId;
|
private readonly SemaphoreSlim _writeLock = new SemaphoreSlim(1, 1);
|
private readonly ConcurrentDictionary<string, TaskCompletionSource<IpcEnvelope>> _pending = new();
|
|
private NamedPipeClientStream? _pipe;
|
private Task? _readLoop;
|
private volatile bool _disposed;
|
private volatile bool _connected;
|
private int _inFlight;
|
|
/// <summary>握手成功后的 Broker 纪元（重启后递增，用于本地陈旧判定）</summary>
|
public long ServerEpoch { get; private set; }
|
|
/// <summary>Broker 生效配置哈希（INV-04 对账参照）</summary>
|
public string BrokerConfigHash { get; private set; } = "";
|
|
public string BrokerId { get; private set; } = "";
|
|
public bool IsConnected => _connected && !_disposed;
|
|
/// <summary>连接/握手默认超时（毫秒）</summary>
|
public int ConnectTimeoutMs { get; set; } = 3000;
|
|
/// <summary>单次 RPC 默认等待上限（毫秒）</summary>
|
public int DefaultRequestTimeoutMs { get; set; } = 10_000;
|
|
public BrokerClientChannel(string pipeName, string clientInstanceId)
|
{
|
_pipeName = pipeName ?? throw new ArgumentNullException(nameof(pipeName));
|
_clientInstanceId = clientInstanceId ?? throw new ArgumentNullException(nameof(clientInstanceId));
|
}
|
|
/// <summary>
|
/// 确保连接并完成 Hello 握手（幂等；断线后可重呼）。
|
/// 协议主版本不匹配时抛出 InvalidOperationException 并拒绝后续请求。
|
/// </summary>
|
public async Task EnsureConnectedAsync(CancellationToken ct)
|
{
|
if (_disposed) throw new ObjectDisposedException(nameof(BrokerClientChannel));
|
if (_connected) return;
|
|
var pipe = new NamedPipeClientStream("localhost", _pipeName,
|
PipeDirection.InOut, PipeOptions.Asynchronous);
|
try
|
{
|
using (var timeoutCts = CancellationTokenSource.CreateLinkedTokenSource(ct))
|
{
|
timeoutCts.CancelAfter(ConnectTimeoutMs);
|
await pipe.ConnectAsync(timeoutCts.Token).ConfigureAwait(false);
|
}
|
}
|
catch (Exception ex)
|
{
|
pipe.Dispose();
|
throw new InvalidOperationException($"BROKER_UNREACHABLE: 无法连接 DeviceBroker 管道 [{_pipeName}]: {ex.Message}", ex);
|
}
|
|
_pipe = pipe;
|
_connected = true;
|
_readLoop = Task.Run(() => ReadLoopAsync(pipe, ct));
|
|
// Hello 握手
|
var hello = NewEnvelope(IpcMessageTypes.Hello);
|
var resp = await SendAsync(hello, ct).ConfigureAwait(false);
|
var result = BrokerIpc.DeserializeBody<HelloResult>(resp.BodyJson);
|
if (result == null)
|
throw new InvalidOperationException("BROKER_HANDSHAKE_FAILED: Hello 响应无效");
|
if (result.ProtocolVersion != BrokerIpc.ProtocolVersion)
|
{
|
_connected = false;
|
throw new InvalidOperationException(
|
$"BROKER_PROTOCOL_MISMATCH: 客户端 v{BrokerIpc.ProtocolVersion}, Broker v{result.ProtocolVersion}");
|
}
|
ServerEpoch = result.ServerEpoch;
|
BrokerConfigHash = result.ConfigHash;
|
BrokerId = result.BrokerId;
|
Log.Info($"[BrokerChannel] 已连接 {_pipeName}: BrokerId={BrokerId}, epoch={ServerEpoch}, configHash={ShortHash(BrokerConfigHash)}");
|
}
|
|
/// <summary>
|
/// 发送一帧请求并等待其关联响应。超时/断线均返回可区分的错误封套（不抛裸异常）。
|
/// </summary>
|
public async Task<IpcEnvelope> SendAsync(IpcEnvelope request, CancellationToken ct)
|
{
|
if (_disposed) throw new ObjectDisposedException(nameof(BrokerClientChannel));
|
var pipe = _pipe;
|
if (!_connected || pipe == null)
|
return ErrorEnvelope(request, "BROKER_UNREACHABLE", "未连接 DeviceBroker");
|
|
var tcs = new TaskCompletionSource<IpcEnvelope>(TaskCreationOptions.RunContinuationsAsynchronously);
|
_pending[request.RequestId] = tcs;
|
Interlocked.Increment(ref _inFlight);
|
try
|
{
|
byte[] frame;
|
try
|
{
|
frame = BrokerIpc.EncodeFrame(BrokerIpc.EncodeEnvelope(request));
|
}
|
catch (Exception ex)
|
{
|
return ErrorEnvelope(request, "PROTOCOL_ERROR", ex.Message);
|
}
|
|
// 写入串行化；pipe 半双工交错下 async 写可能挂死，故写入不加应用级超时（依赖读侧 deadline + 断线感知）
|
await _writeLock.WaitAsync(ct).ConfigureAwait(false);
|
try
|
{
|
await pipe.WriteAsync(frame, 0, frame.Length, ct).ConfigureAwait(false);
|
await pipe.FlushAsync(ct).ConfigureAwait(false);
|
}
|
catch (IOException ex)
|
{
|
Fault($"管道写失败: {ex.Message}");
|
return ErrorEnvelope(request, "BROKER_IO_ERROR", ex.Message);
|
}
|
finally
|
{
|
_writeLock.Release();
|
}
|
|
using (var timeoutCts = CancellationTokenSource.CreateLinkedTokenSource(ct))
|
{
|
int waitMs = request.DeadlineUtcMs > 0
|
? (int)Math.Max(1, request.DeadlineUtcMs - BrokerIpc.NowUtcMs())
|
: DefaultRequestTimeoutMs;
|
var delay = Task.Delay(waitMs, timeoutCts.Token);
|
var winner = await Task.WhenAny(tcs.Task, delay).ConfigureAwait(false);
|
if (winner == tcs.Task)
|
{
|
timeoutCts.Cancel(); // 释放 delay 定时器
|
return await tcs.Task.ConfigureAwait(false);
|
}
|
// INV-06：请求已写出且可能在途 → 结果未知(Unknown)，禁止当未执行
|
return UnknownEnvelope(request, "BROKER_TIMEOUT",
|
"等待响应超时，结果未知，按未完成处理");
|
}
|
}
|
finally
|
{
|
_pending.TryRemove(request.RequestId, out _);
|
Interlocked.Decrement(ref _inFlight);
|
}
|
}
|
|
/// <summary>便捷方法：提交一次 OneShot 设备动作，返回结果体。</summary>
|
public async Task<BrokerIpc.ResultBody> SubmitAsync(BrokerIpc.SubmitBody body, string configHash, CancellationToken ct)
|
{
|
await EnsureConnectedAsync(ct).ConfigureAwait(false);
|
var env = NewEnvelope(IpcMessageTypes.SubmitTask);
|
env.ConfigHash = configHash ?? "";
|
env.BodyJson = BrokerIpc.SerializeBody(body);
|
var resp = await SendAsync(env, ct).ConfigureAwait(false);
|
var result = BrokerIpc.DeserializeBody<BrokerIpc.ResultBody>(resp.BodyJson)
|
?? new BrokerIpc.ResultBody { State = "Unknown", ErrorCode = "BAD_RESPONSE" };
|
return result;
|
}
|
|
// ==================== 内部 ====================
|
|
private async Task ReadLoopAsync(NamedPipeClientStream pipe, CancellationToken ct)
|
{
|
var lenBuf = new byte[4];
|
try
|
{
|
while (!ct.IsCancellationRequested && !_disposed)
|
{
|
await ReadExactAsync(pipe, lenBuf, 4, ct).ConfigureAwait(false);
|
int len = BinaryPrimitives.ReadInt32BigEndian(lenBuf);
|
if (len <= 0 || len > BrokerIpc.MaxFrameBytes)
|
throw new IOException($"非法帧长度 {len}");
|
var payload = new byte[len];
|
await ReadExactAsync(pipe, payload, len, ct).ConfigureAwait(false);
|
|
IpcEnvelope env;
|
try
|
{
|
env = BrokerIpc.DecodeEnvelope(Encoding.UTF8.GetString(payload));
|
}
|
catch (Exception)
|
{
|
continue; // 无法解析的单帧丢弃，不影响后续关联
|
}
|
|
if (!string.IsNullOrEmpty(env.CorrelationId)
|
&& _pending.TryGetValue(env.CorrelationId, out var tcs))
|
{
|
tcs.TrySetResult(env);
|
}
|
}
|
}
|
catch (OperationCanceledException) { /* 正常退出 */ }
|
catch (Exception ex)
|
{
|
if (!_disposed) Fault($"读循环终止: {ex.Message}");
|
}
|
}
|
|
private static async Task ReadExactAsync(Stream s, byte[] buf, int count, CancellationToken ct)
|
{
|
int off = 0;
|
while (off < count)
|
{
|
int n = await s.ReadAsync(buf, off, count - off, ct).ConfigureAwait(false);
|
if (n <= 0) throw new EndOfStreamException("管道已关闭");
|
off += n;
|
}
|
}
|
|
/// <summary>连接失效：置降级态，等待下次 EnsureConnectedAsync 重连；在途请求由其自身超时收口。</summary>
|
private void Fault(string reason)
|
{
|
if (!_connected) return;
|
_connected = false;
|
Log.Warn($"[BrokerChannel] 连接失效({reason})，将在下次请求时重连");
|
try { _pipe?.Dispose(); } catch { }
|
_pipe = null;
|
}
|
|
private IpcEnvelope NewEnvelope(string type) => new IpcEnvelope
|
{
|
ProtocolVersion = BrokerIpc.ProtocolVersion,
|
MessageType = type,
|
RequestId = Guid.NewGuid().ToString("N"),
|
CorrelationId = "",
|
ClientInstanceId = _clientInstanceId,
|
AuthorityId = BrokerId
|
};
|
|
private static IpcEnvelope ErrorEnvelope(IpcEnvelope? request, string code, string message)
|
=> BuildEnvelope(request, "Failed", code, message);
|
|
private static IpcEnvelope UnknownEnvelope(IpcEnvelope? request, string code, string message)
|
=> BuildEnvelope(request, "Unknown", code, message);
|
|
private static IpcEnvelope BuildEnvelope(IpcEnvelope? request, string state, string code, string message)
|
=> new IpcEnvelope
|
{
|
ProtocolVersion = BrokerIpc.ProtocolVersion,
|
MessageType = "Result",
|
CorrelationId = request?.RequestId ?? "",
|
BodyJson = "{\"State\":\"" + state + "\",\"ErrorCode\":\"" + code + "\",\"Readback\":\"" +
|
message.Replace("\"", "'") + "\"}"
|
};
|
|
private static string ShortHash(string h)
|
=> string.IsNullOrEmpty(h) ? "" : (h.Length > 8 ? h.Substring(0, 8) : h);
|
|
public void Dispose()
|
{
|
if (_disposed) return;
|
_disposed = true;
|
_connected = false;
|
try { _pipe?.Dispose(); } catch { }
|
_pipe = null;
|
_writeLock.Dispose();
|
}
|
}
|
}
|
