using System;
|
using System.Configuration;
|
using System.IO;
|
using System.Net.Sockets;
|
using System.Threading;
|
using System.Threading.Tasks;
|
using CanSimulator;
|
using log4net;
|
|
namespace Aging.Infrastructure.Drivers
|
{
|
/// <summary>
|
/// Local 传输后端：TCP 直连 CanSimulatorServer，握手 HELLO 挂载 (bus,nodeIndex)，
|
/// 命令串经 TransceiveAsync 一问一答，接收帧由服务端 EV:RECV 主动推送触发 FramePushed。
|
/// </summary>
|
internal sealed class NodeChannelTransport : ICanTransport
|
{
|
private static readonly ILog Log = LogManager.GetLogger(typeof(NodeChannelTransport));
|
|
private readonly string _host;
|
private readonly int _port;
|
private readonly string _clientTag;
|
|
private TcpClient _socket;
|
private Stream _stream;
|
private CancellationTokenSource _cts;
|
private Task _readLoop;
|
|
// 一问一答：同一时刻仅一个在途请求
|
private readonly SemaphoreSlim _requestLock = new(1, 1);
|
private TaskCompletionSource<string> _pending;
|
|
private volatile bool _connected;
|
private bool _disposed;
|
private string _busName;
|
private int _nodeIndex = -1;
|
|
public bool IsConnected => _connected;
|
public event Action<string> FramePushed;
|
public event Action TransportBroken;
|
|
public NodeChannelTransport(string hostOverride, int? portOverride, string clientTag)
|
{
|
_host = hostOverride ?? ReadCfg("CanSimulator.Host", "127.0.0.1");
|
_port = portOverride ?? ReadIntCfg("CanSimulator.Port", 9000);
|
_clientTag = clientTag ?? $"{Environment.MachineName}/{System.Diagnostics.Process.GetCurrentProcess().Id}";
|
}
|
|
private static string ReadCfg(string key, string def)
|
{
|
try { var v = ConfigurationManager.AppSettings[key]; return string.IsNullOrWhiteSpace(v) ? def : v; }
|
catch { return def; }
|
}
|
private static int ReadIntCfg(string key, int def)
|
=> int.TryParse(ReadCfg(key, null), out int v) ? v : def;
|
|
public async Task ConnectAsync(string busName, int nodeIndex, CancellationToken ct)
|
{
|
if (_disposed) throw new ObjectDisposedException(nameof(NodeChannelTransport));
|
_busName = busName;
|
_nodeIndex = nodeIndex;
|
|
try
|
{
|
_socket = new TcpClient();
|
await _socket.ConnectAsync(_host, _port).ConfigureAwait(false);
|
_socket.NoDelay = true;
|
_stream = _socket.GetStream();
|
_cts = new CancellationTokenSource();
|
|
// HELLO 握手：读循环启动前同步直读一行响应（此时无并发推送）
|
string ack = await HandshakeAsync(
|
$"HELLO:{busName}:{nodeIndex}:{_clientTag}", ct).ConfigureAwait(false);
|
|
if (ack == "ERR:NODE_BUSY")
|
throw new InvalidOperationException(
|
$"模拟器总线 {busName} 上节点编号 {nodeIndex} 已被占用，请修改设备端点的节点编号");
|
if (!ack.StartsWith("OK:HELLO", StringComparison.Ordinal))
|
throw new InvalidOperationException($"模拟器握手失败: {ack}");
|
|
_connected = true;
|
_readLoop = Task.Run(() => ReadLoopAsync(_cts.Token));
|
Log.Info($"[CAN-TCP] 已挂载 {busName}:N{nodeIndex} @ {_host}:{_port}");
|
}
|
catch (SocketException ex)
|
{
|
SafeCleanupSocket();
|
throw new InvalidOperationException(
|
$"无法连接 CAN 模拟器服务 {_host}:{_port}（{ex.Message}）。请先启动 CanSimulatorServer.exe", ex);
|
}
|
}
|
|
public async Task DisconnectAsync()
|
{
|
if (!_connected) return;
|
try { await RequestLineAsync("DISCONNECT", CancellationToken.None).ConfigureAwait(false); }
|
catch { /* 断开尽力而为 */ }
|
Teardown();
|
}
|
|
public async Task<string> TransceiveAsync(string line, CancellationToken ct)
|
{
|
if (_disposed) throw new ObjectDisposedException(nameof(NodeChannelTransport));
|
if (!_connected) throw new InvalidOperationException("CAN 通道未连接");
|
try
|
{
|
return await RequestLineAsync(line, ct).ConfigureAwait(false);
|
}
|
catch (IOException ex)
|
{
|
OnBroken();
|
throw new InvalidOperationException("CAN 模拟器连接已中断: " + ex.Message, ex);
|
}
|
}
|
|
/// <summary>握手专用：写 HELLO 后直接同步读取一行响应（读循环尚未启动，无并发）。</summary>
|
private async Task<string> HandshakeAsync(string helloLine, CancellationToken ct)
|
{
|
byte[] frame = SimulatorFraming.Encode(helloLine);
|
await _stream.WriteAsync(frame, 0, frame.Length, ct).ConfigureAwait(false);
|
await _stream.FlushAsync(ct).ConfigureAwait(false);
|
string resp = await SimulatorFraming.ReadFrameAsync(_stream, ct).ConfigureAwait(false);
|
return resp ?? "<EOF>";
|
}
|
|
/// <summary>发送一行并等待其响应行；EV:RECV 推送在此过程中被分流出去，不计为响应。</summary>
|
private async Task<string> RequestLineAsync(string line, CancellationToken ct)
|
{
|
await _requestLock.WaitAsync(ct).ConfigureAwait(false);
|
try
|
{
|
var tcs = new TaskCompletionSource<string>(TaskCreationOptions.RunContinuationsAsynchronously);
|
_pending = tcs;
|
|
byte[] frame = SimulatorFraming.Encode(line);
|
await _stream.WriteAsync(frame, 0, frame.Length, ct).ConfigureAwait(false);
|
await _stream.FlushAsync(ct).ConfigureAwait(false);
|
|
// deadline：默认 5s，超时视为结果未知（上层禁止假定未执行）
|
var timeoutTask = Task.Delay(TimeSpan.FromSeconds(5), ct);
|
var done = await Task.WhenAny(tcs.Task, timeoutTask).ConfigureAwait(false);
|
if (done != tcs.Task)
|
{
|
_pending = null;
|
throw new TimeoutException($"CAN 模拟器命令超时: {line}");
|
}
|
_pending = null;
|
ct.ThrowIfCancellationRequested();
|
return await tcs.Task.ConfigureAwait(false);
|
}
|
finally
|
{
|
_requestLock.Release();
|
}
|
}
|
|
private async Task ReadLoopAsync(CancellationToken ct)
|
{
|
try
|
{
|
while (!ct.IsCancellationRequested)
|
{
|
string resp = await SimulatorFraming.ReadFrameAsync(_stream, ct).ConfigureAwait(false);
|
if (resp == null) break; // EOF
|
|
if (resp.StartsWith("EV:", StringComparison.Ordinal))
|
{
|
FramePushed?.Invoke(resp);
|
continue;
|
}
|
|
// 普通响应：交给在途请求
|
_pending?.TrySetResult(resp);
|
}
|
}
|
catch (IOException) { }
|
catch (InvalidDataException) { }
|
catch (OperationCanceledException) { }
|
catch (Exception ex) { Log.Warn($"[CAN-TCP] 读循环异常: {ex.Message}"); }
|
finally
|
{
|
if (!_disposed) OnBroken();
|
}
|
}
|
|
private void OnBroken()
|
{
|
if (!_connected && _pending == null) return;
|
_connected = false;
|
_pending?.TrySetException(new IOException("CAN 模拟器通道已断开"));
|
TransportBroken?.Invoke();
|
}
|
|
private void Teardown()
|
{
|
_connected = false;
|
try { _cts?.Cancel(); } catch { }
|
SafeCleanupSocket();
|
}
|
|
private void SafeCleanupSocket()
|
{
|
try { _stream?.Dispose(); } catch { }
|
try { _socket?.Close(); } catch { }
|
_stream = null;
|
_socket = null;
|
}
|
|
public void Dispose()
|
{
|
if (_disposed) return;
|
_disposed = true;
|
Teardown();
|
try { _readLoop?.Wait(200); } catch { }
|
_requestLock.Dispose();
|
_cts?.Dispose();
|
}
|
}
|
}
|
