using System;
|
using System.IO;
|
using System.Net.Sockets;
|
using System.Threading;
|
using System.Threading.Tasks;
|
using CanSimulator;
|
using log4net;
|
|
namespace CanSimulatorServer
|
{
|
/// <summary>
|
/// 单连接会话：HELLO 状态机 + 命令执行（保序响应） + 帧推送（写锁串行化保序）。
|
/// 读循环单线程处理请求；响应与推送经同一写锁发出，客户端按到达顺序分流。
|
/// </summary>
|
public sealed class ServerSession : IDisposable
|
{
|
private static readonly ILog Log = LogManager.GetLogger(typeof(ServerSession));
|
|
private readonly TcpClient _client;
|
private readonly CanBusManager _manager;
|
private readonly Action<ServerSession> _onClosed;
|
private readonly CancellationTokenSource _cts = new();
|
private readonly SemaphoreSlim _writeLock = new(1, 1);
|
|
private string? _busName;
|
private CanNode? _node;
|
private CanCommandExecutor? _executor;
|
private long _droppedPushes;
|
|
/// <summary>推送暂存门：命令处理期间路由产生的帧先入队，响应写完后再冲，
|
/// 保证客户端“首行必为响应、推送随后”的保序契约</summary>
|
private readonly object _pushGate = new();
|
private bool _inCommand;
|
private readonly System.Collections.Generic.List<string> _heldPushes = new();
|
private const int HeldPushCapacity = 1000;
|
|
public string ClientTag { get; private set; } = "";
|
public int NodeIndex { get; private set; } = -1;
|
public bool Attached => _node != null;
|
public long DroppedPushes => Interlocked.Read(ref _droppedPushes);
|
|
/// <summary>节点帧到达事件（UI 帧表订阅）</summary>
|
public event Action<string, string, int>? FramePushed; // (bus, pushLine, nodeIndex)
|
|
/// <summary>总线观察级帧事件（路由发生后向 UI 报告，由宿主挂接）</summary>
|
internal ServerSession(TcpClient client, CanBusManager manager, Action<ServerSession> onClosed)
|
{
|
_client = client;
|
_manager = manager;
|
_onClosed = onClosed;
|
}
|
|
public void Start()
|
{
|
_ = Task.Run(() => RunAsync(_cts.Token));
|
}
|
|
private async Task RunAsync(CancellationToken ct)
|
{
|
Stream? stream = null;
|
try
|
{
|
stream = _client.GetStream();
|
while (!ct.IsCancellationRequested)
|
{
|
string? line = await SimulatorFraming.ReadFrameAsync(stream, ct).ConfigureAwait(false);
|
if (line == null) break; // EOF / 截断 → 会话结束
|
|
string? response = Process(line);
|
if (response != null)
|
{
|
await WriteAsync(response, ct).ConfigureAwait(false);
|
await FlushHeldPushesAsync(ct).ConfigureAwait(false); // 响应先行，补发处理期间暂存的帧
|
}
|
}
|
}
|
catch (InvalidDataException)
|
{
|
// 协议失步：断开该会话，宿主继续监听
|
}
|
catch (IOException) { }
|
catch (OperationCanceledException) { }
|
catch (Exception ex)
|
{
|
Log.Warn($"[SimServer] 会话异常结束: {ex.Message}");
|
}
|
finally
|
{
|
Close();
|
}
|
}
|
|
/// <summary>处理一行请求，返回响应文本（无需响应时返回 null）。</summary>
|
private string? Process(string line)
|
{
|
if (line.StartsWith("HELLO:", StringComparison.OrdinalIgnoreCase))
|
return HandleHello(line);
|
|
if (line.StartsWith("DISCONNECT", StringComparison.OrdinalIgnoreCase))
|
{
|
// 显式断开：回 OK 后由 EOF/关闭收尾
|
return "OK:DISCONNECTED";
|
}
|
|
if (_node == null)
|
return "ERR:NOT_HANDSHAKED";
|
|
if (line.StartsWith("SEND:", StringComparison.OrdinalIgnoreCase) ||
|
line.StartsWith("QUERY:", StringComparison.OrdinalIgnoreCase))
|
{
|
string resp;
|
lock (_pushGate) { _inCommand = true; } // 处理期间的推送改走暂存
|
try
|
{
|
resp = _executor!.Execute(line);
|
}
|
finally
|
{
|
lock (_pushGate) { _inCommand = false; }
|
}
|
// 裸数据（非 OK/ERR 前缀）补 OK:
|
if (!resp.StartsWith("OK", StringComparison.Ordinal) &&
|
!resp.StartsWith("ERR", StringComparison.Ordinal))
|
resp = "OK:" + resp;
|
return resp;
|
}
|
|
return $"ERR:UNKNOWN_CMD:{line}";
|
}
|
|
private string? HandleHello(string line)
|
{
|
if (_node != null)
|
return "ERR:ALREADY_HELLO";
|
|
var parts = line.Split(':'); // HELLO:bus:node:tag
|
if (parts.Length < 4 || !int.TryParse(parts[2], out int nodeIndex) || nodeIndex < 0)
|
return "ERR:BAD_FRAME:HELLO 格式应为 HELLO:bus:nodeIndex:tag";
|
|
string bus = parts[1];
|
ClientTag = parts[3];
|
bool ok = _manager.TryAttach(bus, nodeIndex, ClientTag, OnNodeFrame, out var node, out var busInst);
|
if (!ok) return "ERR:NODE_BUSY";
|
|
_node = node;
|
_busName = bus;
|
NodeIndex = nodeIndex;
|
_executor = new CanCommandExecutor(node!, busInst!);
|
return $"OK:HELLO:{bus}:{nodeIndex}";
|
}
|
|
/// <summary>总线路由到本节点的帧 → 推送（命令处理中暂存，否则即发即走）</summary>
|
private void OnNodeFrame(CanFrame frame)
|
{
|
string push = CanCommandExecutor.FormatPush(frame);
|
FramePushed?.Invoke(_busName ?? "?", push, NodeIndex);
|
|
lock (_pushGate)
|
{
|
if (_inCommand)
|
{
|
if (_heldPushes.Count >= HeldPushCapacity)
|
{
|
_heldPushes.RemoveAt(0); // 满则丢最老，由客户端 RECV_ALL 兜底补账
|
Interlocked.Increment(ref _droppedPushes);
|
}
|
_heldPushes.Add(push);
|
return;
|
}
|
}
|
|
// 空闲期：即发即走，写锁串行化保帧序；失败计入丢帧
|
_ = WriteAsync(push, CancellationToken.None).ContinueWith(t =>
|
{
|
if (t.IsFaulted) Interlocked.Increment(ref _droppedPushes);
|
}, TaskScheduler.Default);
|
}
|
|
private async Task FlushHeldPushesAsync(CancellationToken ct)
|
{
|
string[] pending;
|
lock (_pushGate)
|
{
|
if (_heldPushes.Count == 0) return;
|
pending = _heldPushes.ToArray();
|
_heldPushes.Clear();
|
}
|
foreach (var p in pending)
|
await WriteAsync(p, ct).ConfigureAwait(false);
|
}
|
|
private async Task WriteAsync(string text, CancellationToken ct)
|
{
|
byte[] frame = SimulatorFraming.Encode(text);
|
await _writeLock.WaitAsync(ct).ConfigureAwait(false);
|
try
|
{
|
if (_client.Connected)
|
{
|
var s = _client.GetStream();
|
await s.WriteAsync(frame, 0, frame.Length, ct).ConfigureAwait(false);
|
await s.FlushAsync(ct).ConfigureAwait(false);
|
}
|
}
|
finally
|
{
|
_writeLock.Release();
|
}
|
}
|
|
public void Close()
|
{
|
try
|
{
|
if (_node != null && _busName != null)
|
_manager.Detach(_busName, _node);
|
_node = null;
|
_cts.Cancel();
|
_client.Close();
|
}
|
catch (Exception ex)
|
{
|
Log.Debug($"[SimServer] 会话关闭异常: {ex.Message}");
|
}
|
finally
|
{
|
_onClosed(this);
|
}
|
}
|
|
public void Dispose()
|
{
|
Close();
|
_cts.Dispose();
|
_writeLock.Dispose();
|
}
|
}
|
}
|
