using DeviceService.Core;
|
using System.IO;
|
using System.IO.Pipes;
|
using System.Text.Json;
|
|
namespace DeviceService.Service.NamePipe
|
{
|
/// <summary>
|
/// 基于命名管道的服务端实现（Message 模式）。
|
/// 启动后持续监听客户端连接，收到的每条消息通过 <see cref="DataReceived"/> 事件上抛，
|
/// 并携带可回写数据的 <see cref="NamePipeClient"/> 实例。
|
/// </summary>
|
public class NamePipeService : IService, IDisposable
|
{
|
/// <summary>默认允许的最大管道实例数（即最大并发连接数）。</summary>
|
public const int DefaultMaxInstances = 254;
|
|
/// <summary>停止服务时等待后台任务结束的最长时间。</summary>
|
private static readonly TimeSpan StopTimeout = TimeSpan.FromSeconds(3);
|
|
private readonly string _pipeName;
|
private readonly int _maxInstances;
|
private readonly object _sync = new object();
|
private readonly Dictionary<NamePipeClient, Task> _clients = new Dictionary<NamePipeClient, Task>();
|
|
private CancellationTokenSource? _cts;
|
private NamedPipeServerStream? _listenStream;
|
private Task? _acceptTask;
|
private bool _running;
|
private bool _disposed;
|
|
/// <param name="pipeName">管道名称，为空时在 Start 阶段校验失败。</param>
|
/// <param name="maxInstances">最大并发连接数。</param>
|
public NamePipeService(string? pipeName = null, int maxInstances = DefaultMaxInstances)
|
{
|
if (maxInstances <= 0)
|
{
|
throw new ArgumentOutOfRangeException(nameof(maxInstances), "最大实例数必须大于 0。");
|
}
|
_pipeName = pipeName ?? string.Empty;
|
_maxInstances = maxInstances;
|
}
|
public event EventHandler<IClient>? ClientConnected;
|
public event EventHandler<IClient>? ClientDisConnected;
|
public event EventHandler<RequestArg>? DataReceived;
|
|
/// <summary>管道名称。</summary>
|
public string PipeName => _pipeName;
|
|
/// <summary>服务是否处于监听状态。</summary>
|
public bool IsRunning
|
{
|
get
|
{
|
lock (_sync)
|
{
|
return _running;
|
}
|
}
|
}
|
|
/// <summary>当前已连接的客户端数量。</summary>
|
public int ClientCount
|
{
|
get
|
{
|
lock (_sync)
|
{
|
return _clients.Count;
|
}
|
}
|
}
|
|
public bool Start(out string msg)
|
{
|
lock (_sync)
|
{
|
if (_disposed)
|
{
|
msg = "服务已释放，无法启动。";
|
return false;
|
}
|
|
if (_running)
|
{
|
msg = "服务已处于运行状态。";
|
return false;
|
}
|
|
if (string.IsNullOrWhiteSpace(_pipeName))
|
{
|
msg = "管道名称未设置，无法启动。";
|
return false;
|
}
|
|
NamedPipeServerStream listenStream;
|
try
|
{
|
listenStream = CreatePipeStream();
|
}
|
catch (Exception ex)
|
{
|
msg = $"创建命名管道 \"{_pipeName}\" 失败：{ex.Message}";
|
return false;
|
}
|
|
var cts = new CancellationTokenSource();
|
_cts = cts;
|
_listenStream = listenStream;
|
_running = true;
|
_acceptTask = Task.Run(() => AcceptLoopAsync(listenStream, cts.Token));
|
}
|
|
msg = $"命名管道 \"{_pipeName}\" 已启动监听。";
|
return true;
|
}
|
|
public bool Stop(out string msg)
|
{
|
Task? acceptTask;
|
List<Task> clientTasks = new List<Task>();
|
|
lock (_sync)
|
{
|
if (!_running)
|
{
|
msg = "服务未运行。";
|
return false;
|
}
|
|
_running = false;
|
acceptTask = _acceptTask;
|
_acceptTask = null;
|
|
try
|
{
|
_cts?.Cancel();
|
}
|
catch
|
{
|
// 忽略取消过程中的异常
|
}
|
|
try
|
{
|
_listenStream?.Dispose();
|
}
|
catch
|
{
|
// 忽略关闭监听流的异常
|
}
|
|
_listenStream = null;
|
|
// 关闭所有客户端连接，使阻塞中的读取立即结束
|
foreach (KeyValuePair<NamePipeClient, Task> item in _clients)
|
{
|
try
|
{
|
item.Key.Dispose();
|
}
|
catch
|
{
|
// 忽略单个连接的关闭异常
|
}
|
|
clientTasks.Add(item.Value);
|
}
|
|
_clients.Clear();
|
|
_cts?.Dispose();
|
_cts = null;
|
}
|
|
try
|
{
|
acceptTask?.Wait(StopTimeout);
|
if (clientTasks.Count > 0)
|
{
|
Task.WaitAll(clientTasks.ToArray(), StopTimeout);
|
}
|
}
|
catch (AggregateException)
|
{
|
// 后台任务在校验/取消阶段抛出的异常无需向上传播
|
}
|
|
msg = "服务已停止。";
|
return true;
|
}
|
|
private NamedPipeServerStream CreatePipeStream()
|
{
|
var pipe = new NamedPipeServerStream(
|
_pipeName,
|
PipeDirection.InOut,
|
_maxInstances,
|
PipeTransmissionMode.Message,
|
PipeOptions.Asynchronous);
|
pipe.ReadMode = PipeTransmissionMode.Message;
|
return pipe;
|
}
|
|
/// <summary>
|
/// 连接接受循环：每接受一个连接就交给独立的处理任务，并准备下一个管道实例继续监听。
|
/// </summary>
|
private async Task AcceptLoopAsync(NamedPipeServerStream initial, CancellationToken token)
|
{
|
NamedPipeServerStream? listenStream = initial;
|
try
|
{
|
while (!token.IsCancellationRequested)
|
{
|
NamedPipeServerStream stream = listenStream!;
|
listenStream = null;
|
|
try
|
{
|
await stream.WaitForConnectionAsync(token).ConfigureAwait(false);
|
}
|
catch (OperationCanceledException)
|
{
|
stream.Dispose();
|
break;
|
}
|
catch (ObjectDisposedException)
|
{
|
break;
|
}
|
catch (IOException)
|
{
|
stream.Dispose();
|
break;
|
}
|
|
// .NET Framework 下 WaitForConnectionAsync 可能提前返回，此处校验连接确实建立
|
if (!stream.IsConnected)
|
{
|
try
|
{
|
await Task.Delay(50, token).ConfigureAwait(false);
|
}
|
catch (OperationCanceledException)
|
{
|
break;
|
}
|
|
listenStream = stream;
|
continue;
|
}
|
|
var client = new NamePipeClient(stream);
|
Task handler = Task.Run(() => HandleClientAsync(client, token));
|
RegisterClient(client, handler);
|
|
// 为下一个连接准备新的管道实例
|
try
|
{
|
listenStream = CreatePipeStream();
|
}
|
catch (Exception)
|
{
|
break;
|
}
|
|
lock (_sync)
|
{
|
if (!_running)
|
{
|
break;
|
}
|
|
_listenStream = listenStream;
|
}
|
}
|
}
|
catch (Exception)
|
{
|
// 监听循环中的未知异常直接结束循环，由 Stop 负责收尾
|
}
|
finally
|
{
|
try
|
{
|
listenStream?.Dispose();
|
}
|
catch
|
{
|
// 忽略关闭异常
|
}
|
|
lock (_sync)
|
{
|
_listenStream = null;
|
}
|
}
|
}
|
|
/// <summary>
|
/// 单个连接的读取循环：按行读取消息并触发 <see cref="DataReceived"/>。
|
/// </summary>
|
private async Task HandleClientAsync(NamePipeClient client, CancellationToken token)
|
{
|
OnClientEvent(ClientConnected, client);
|
try
|
{
|
while (!token.IsCancellationRequested)
|
{
|
string? message = await client.ReadMessageAsync().ConfigureAwait(false);
|
if (message == null)
|
{
|
break;
|
}
|
|
RaiseDataReceived(client, message);
|
}
|
}
|
catch (OperationCanceledException)
|
{
|
// 服务停止导致的正常取消
|
}
|
catch (Exception)
|
{
|
// 连接断开、数据非法等情况直接结束该连接的读取
|
}
|
finally
|
{
|
OnClientEvent(ClientDisConnected, client);
|
UnregisterClient(client);
|
client.Dispose();
|
}
|
}
|
|
private void OnClientEvent(EventHandler<IClient>? handler, IClient client)
|
{
|
if (handler == null)
|
{
|
return;
|
}
|
|
try
|
{
|
handler(this, client);
|
}
|
catch
|
{
|
// 订阅方的异常不应影响连接的建立与释放
|
}
|
}
|
|
private void RaiseDataReceived(NamePipeClient client, string raw)
|
{
|
var data = JsonSerializer.Deserialize<RequestData>(raw);
|
var arg = new RequestArg
|
{
|
Data = data,
|
Raw = raw,
|
Client = client
|
};
|
OnDataReceived(arg);
|
}
|
|
private void RegisterClient(NamePipeClient client, Task handler)
|
{
|
lock (_sync)
|
{
|
if (_running)
|
{
|
_clients[client] = handler;
|
return;
|
}
|
}
|
|
// 服务正在停止，直接释放该连接
|
client.Dispose();
|
}
|
|
private void UnregisterClient(NamePipeClient client)
|
{
|
lock (_sync)
|
{
|
_clients.Remove(client);
|
}
|
}
|
|
private void OnDataReceived(RequestArg arg)
|
{
|
EventHandler<RequestArg>? handler = DataReceived;
|
if (handler == null)
|
{
|
return;
|
}
|
|
try
|
{
|
handler?.Invoke(this, arg);
|
}
|
catch
|
{
|
// 订阅方的异常不应中断管道读取
|
}
|
}
|
|
public void Dispose()
|
{
|
if (_disposed)
|
{
|
return;
|
}
|
|
try
|
{
|
Stop(out _);
|
}
|
catch
|
{
|
// 释放阶段的异常忽略
|
}
|
|
_disposed = true;
|
}
|
}
|
}
|
