using System;
|
using System.Collections.Concurrent;
|
using System.IO;
|
using System.IO.Pipes;
|
using System.Text;
|
using System.Threading;
|
using System.Threading.Tasks;
|
|
namespace ConnectionUtil
|
{
|
public class PipeServer : IDisposable
|
{
|
private readonly string _pipeName;
|
private readonly int _maxNumberOfServerInstances;
|
private readonly CancellationTokenSource _cancellationTokenSource;
|
private readonly ConcurrentDictionary<string, NamedPipeServerStream> _activeConnections;
|
private bool _isDisposed;
|
|
// 事件定义
|
public event EventHandler<PipeDataEventArgs> OnData;
|
public event EventHandler<PipeConnectionEventArgs> OnNewConnection;
|
public event EventHandler<PipeConnectionEventArgs> OnClientDisconnected;
|
public event EventHandler<PipeErrorEventArgs> OnError;
|
|
public PipeServer(string pipeName, int maxNumberOfServerInstances = NamedPipeServerStream.MaxAllowedServerInstances)
|
{
|
_pipeName = pipeName;
|
_maxNumberOfServerInstances = maxNumberOfServerInstances;
|
_cancellationTokenSource = new CancellationTokenSource();
|
_activeConnections = new ConcurrentDictionary<string, NamedPipeServerStream>();
|
}
|
|
public void Start()
|
{
|
Task.Run(() =>
|
{
|
try
|
{
|
while (!_cancellationTokenSource.Token.IsCancellationRequested)
|
{
|
var pipeServer = new NamedPipeServerStream(_pipeName, PipeDirection.InOut,
|
_maxNumberOfServerInstances, PipeTransmissionMode.Message);
|
try
|
{
|
pipeServer.WaitForConnection();
|
|
// 生成唯一的连接ID
|
string connectionId = Guid.NewGuid().ToString();
|
_activeConnections.TryAdd(connectionId, pipeServer);
|
|
// 触发新连接事件
|
OnNewConnection?.Invoke(this, new PipeConnectionEventArgs(connectionId));
|
|
// 开始处理该连接的数据
|
HandleClient(connectionId, pipeServer);
|
}
|
catch (OperationCanceledException)
|
{
|
pipeServer.Dispose();
|
break;
|
}
|
catch (Exception ex)
|
{
|
pipeServer.Dispose();
|
OnError?.Invoke(this, new PipeErrorEventArgs("等待连接时发生错误", ex));
|
}
|
}
|
}
|
catch (Exception ex)
|
{
|
OnError?.Invoke(this, new PipeErrorEventArgs("服务器运行时发生错误", ex));
|
}
|
|
});
|
}
|
|
static string ReadMessage(PipeStream s)
|
{
|
MemoryStream ms = new MemoryStream();
|
byte[] buffer = new byte[0x1000]; // Read in 4 KB blocks
|
do
|
{
|
ms.Write(buffer, 0, s.Read(buffer, 0, buffer.Length));
|
}
|
while (!s.IsMessageComplete);
|
return Encoding.UTF8.GetString(ms.ToArray());
|
}
|
|
private void HandleClient(string connectionId, NamedPipeServerStream pipeServer)
|
{
|
Task.Run(() =>
|
{
|
try
|
{
|
while (pipeServer.IsConnected && !_cancellationTokenSource.Token.IsCancellationRequested)
|
{
|
try
|
{
|
string message = ReadMessage(pipeServer);
|
if (string.IsNullOrEmpty(message))
|
continue;
|
Console.WriteLine(message);
|
if (message == "ping")
|
{
|
message = "pong";
|
}
|
else
|
{
|
message += "Response";
|
}
|
SendMessage(connectionId, message);
|
}
|
catch (Exception ex)
|
{
|
OnError?.Invoke(this, new PipeErrorEventArgs($"处理客户端 {connectionId} 数据时发生错误", ex));
|
break;
|
}
|
}
|
}
|
finally
|
{
|
// 清理连接
|
_activeConnections.TryRemove(connectionId, out _);
|
pipeServer.Dispose();
|
|
// 触发断开连接事件
|
OnClientDisconnected?.Invoke(this, new PipeConnectionEventArgs(connectionId));
|
}
|
});
|
}
|
|
public void SendMessage(string connectionId, string message)
|
{
|
if (_activeConnections.TryGetValue(connectionId, out var pipeServer))
|
{
|
try
|
{
|
var msg = Encoding.UTF8.GetBytes(message);
|
pipeServer.Write(msg, 0, msg.Length);
|
}
|
catch (Exception ex)
|
{
|
OnError?.Invoke(this, new PipeErrorEventArgs($"向客户端 {connectionId} 发送消息时发生错误", ex));
|
}
|
}
|
}
|
|
|
public async Task StartAsync()
|
{
|
try
|
{
|
while (!_cancellationTokenSource.Token.IsCancellationRequested)
|
{
|
var pipeServer = new NamedPipeServerStream(_pipeName, PipeDirection.InOut,
|
_maxNumberOfServerInstances, PipeTransmissionMode.Message);
|
|
try
|
{
|
await pipeServer.WaitForConnectionAsync(_cancellationTokenSource.Token);
|
|
// 生成唯一的连接ID
|
string connectionId = Guid.NewGuid().ToString();
|
_activeConnections.TryAdd(connectionId, pipeServer);
|
|
// 触发新连接事件
|
OnNewConnection?.Invoke(this, new PipeConnectionEventArgs(connectionId));
|
|
// 开始处理该连接的数据
|
_ = HandleClientAsync(connectionId, pipeServer);
|
}
|
catch (OperationCanceledException)
|
{
|
pipeServer.Dispose();
|
break;
|
}
|
catch (Exception ex)
|
{
|
pipeServer.Dispose();
|
OnError?.Invoke(this, new PipeErrorEventArgs("等待连接时发生错误", ex));
|
}
|
}
|
}
|
catch (Exception ex)
|
{
|
OnError?.Invoke(this, new PipeErrorEventArgs("服务器运行时发生错误", ex));
|
}
|
}
|
|
private async Task HandleClientAsync(string connectionId, NamedPipeServerStream pipeServer)
|
{
|
try
|
{
|
using (var reader = new StreamReader(pipeServer))
|
using (var writer = new StreamWriter(pipeServer))
|
{
|
writer.AutoFlush = true;
|
|
while (pipeServer.IsConnected && !_cancellationTokenSource.Token.IsCancellationRequested)
|
{
|
try
|
{
|
string message = await reader.ReadLineAsync();
|
if (string.IsNullOrEmpty(message))
|
continue;
|
|
// 触发数据接收事件
|
OnData?.Invoke(this, new PipeDataEventArgs(connectionId, message));
|
|
// 发送响应
|
await writer.WriteLineAsync($"服务器已收到消息: {message}");
|
}
|
catch (Exception ex)
|
{
|
OnError?.Invoke(this, new PipeErrorEventArgs($"处理客户端 {connectionId} 数据时发生错误", ex));
|
break;
|
}
|
}
|
}
|
}
|
finally
|
{
|
// 清理连接
|
_activeConnections.TryRemove(connectionId, out _);
|
pipeServer.Dispose();
|
|
// 触发断开连接事件
|
OnClientDisconnected?.Invoke(this, new PipeConnectionEventArgs(connectionId));
|
}
|
}
|
|
public async Task SendMessageAsync(string connectionId, string message)
|
{
|
if (_activeConnections.TryGetValue(connectionId, out var pipeServer))
|
{
|
try
|
{
|
using (var writer = new StreamWriter(pipeServer))
|
{
|
writer.AutoFlush = true;
|
await writer.WriteLineAsync(message);
|
}
|
}
|
catch (Exception ex)
|
{
|
OnError?.Invoke(this, new PipeErrorEventArgs($"向客户端 {connectionId} 发送消息时发生错误", ex));
|
}
|
}
|
}
|
|
public void Stop()
|
{
|
_cancellationTokenSource.Cancel();
|
}
|
|
public void Dispose()
|
{
|
if (!_isDisposed)
|
{
|
Stop();
|
foreach (var connection in _activeConnections.Values)
|
{
|
connection.Dispose();
|
}
|
_activeConnections.Clear();
|
_cancellationTokenSource.Dispose();
|
_isDisposed = true;
|
}
|
}
|
}
|
|
public class PipeDataEventArgs : EventArgs
|
{
|
public string ConnectionId { get; }
|
public string Data { get; }
|
|
public PipeDataEventArgs(string connectionId, string data)
|
{
|
ConnectionId = connectionId;
|
Data = data;
|
}
|
}
|
|
public class PipeConnectionEventArgs : EventArgs
|
{
|
public string ConnectionId { get; }
|
|
public PipeConnectionEventArgs(string connectionId)
|
{
|
ConnectionId = connectionId;
|
}
|
}
|
|
public class PipeErrorEventArgs : EventArgs
|
{
|
public string Message { get; }
|
public Exception Exception { get; }
|
|
public PipeErrorEventArgs(string message, Exception exception)
|
{
|
Message = message;
|
Exception = exception;
|
}
|
}
|
|
public class StreamString
|
{
|
private Stream ioStream;
|
private UnicodeEncoding streamEncoding;
|
|
public StreamString(Stream ioStream)
|
{
|
this.ioStream = ioStream;
|
streamEncoding = new UnicodeEncoding();
|
}
|
|
public string ReadString()
|
{
|
int len = 0;
|
|
|
len = ioStream.ReadByte() * 256;
|
len += ioStream.ReadByte();
|
byte[] inBuffer = new byte[len];
|
ioStream.Read(inBuffer, 0, len);
|
|
return streamEncoding.GetString(inBuffer);
|
}
|
|
public int WriteString(string outString)
|
{
|
byte[] outBuffer = streamEncoding.GetBytes(outString);
|
int len = outBuffer.Length;
|
if (len > UInt16.MaxValue)
|
{
|
len = (int)UInt16.MaxValue;
|
}
|
ioStream.WriteByte((byte)(len / 256));
|
ioStream.WriteByte((byte)(len & 255));
|
ioStream.Write(outBuffer, 0, len);
|
ioStream.Flush();
|
|
return outBuffer.Length + 2;
|
}
|
}
|
|
}
|
