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 static class PipeStreamUtil
|
{
|
public static string ReadMessage(this PipeStream ps)
|
{
|
|
MemoryStream ms = new MemoryStream();
|
byte[] buffer = new byte[0x1000];
|
do
|
{
|
ms.Write(buffer, 0, ps.Read(buffer, 0, buffer.Length));
|
}
|
while (!ps.IsMessageComplete);
|
return Encoding.UTF8.GetString(ms.ToArray());
|
}
|
|
public static void WriteMessage(this PipeStream ps, string message)
|
{
|
var msg = Encoding.UTF8.GetBytes(message);
|
ps.Write(msg, 0, msg.Length);
|
ps.Flush();
|
}
|
}
|
|
public class PipeServer2 : IDisposable
|
{
|
private readonly string _pipeName;
|
private readonly int _maxNumberOfServerInstances;
|
private readonly CancellationTokenSource _cancellationTokenSource;
|
private readonly ConcurrentDictionary<string, NamedPipeServerStream> _activeConnections;
|
private readonly ReaderWriterLockSlim _lock;
|
private bool _isDisposed;
|
private bool _isRunning;
|
|
// 事件定义
|
public event EventHandler<PipeDataEventArgs> OnData;
|
public event EventHandler<PipeConnectionEventArgs> OnClientConnected;
|
public event EventHandler<PipeConnectionEventArgs> OnClientDisconnected;
|
public event EventHandler<PipeErrorEventArgs> OnError;
|
|
public PipeServer2(string pipeName, int maxNumberOfServerInstances = 254)
|
{
|
_pipeName = pipeName;
|
_maxNumberOfServerInstances = maxNumberOfServerInstances;
|
_cancellationTokenSource = new CancellationTokenSource();
|
_activeConnections = new ConcurrentDictionary<string, NamedPipeServerStream>();
|
_lock = new ReaderWriterLockSlim();
|
}
|
|
public void Start()
|
{
|
if (_isRunning)
|
return;
|
|
_isRunning = true;
|
|
Task.Run(() =>
|
{
|
while (_isRunning && !_cancellationTokenSource.Token.IsCancellationRequested)
|
{
|
try
|
{
|
var pipeServer = new NamedPipeServerStream(_pipeName, PipeDirection.InOut, _maxNumberOfServerInstances
|
, PipeTransmissionMode.Message);
|
pipeServer.WaitForConnection();
|
|
string clientId = Guid.NewGuid().ToString();
|
if (_activeConnections.TryAdd(clientId, pipeServer))
|
{
|
// 启动处理客户端消息的线程
|
Thread clientThread = new Thread(() => HandleClient(clientId, pipeServer));
|
clientThread.Start();
|
|
OnClientConnected?.Invoke(this, new PipeConnectionEventArgs(clientId));
|
}
|
}
|
catch (Exception ex)
|
{
|
OnError?.Invoke(this, new PipeErrorEventArgs("Connect", ex));
|
Thread.Sleep(1000); // 发生错误时等待一秒再重试
|
}
|
}
|
});
|
}
|
|
|
|
private void HandleClient(string clientId, NamedPipeServerStream pipeServer)
|
{
|
try
|
{
|
while (!_cancellationTokenSource.Token.IsCancellationRequested)
|
{
|
try
|
{
|
// 等待客户端发送消息
|
string message = pipeServer.ReadMessage();
|
if (string.IsNullOrEmpty(message))
|
break;
|
|
OnData?.Invoke(this, new PipeDataEventArgs(clientId, message));
|
}
|
catch (Exception ex)
|
{
|
OnError?.Invoke(this, new PipeErrorEventArgs("HandleClientMessageError", ex));
|
break;
|
}
|
}
|
}
|
finally
|
{
|
if (_activeConnections.TryRemove(clientId, out var pipe))
|
{
|
pipe.Dispose();
|
}
|
OnClientDisconnected?.Invoke(this, new PipeConnectionEventArgs(clientId));
|
}
|
}
|
|
public void SendMessage(string clientId, string message)
|
{
|
if (!_activeConnections.TryGetValue(clientId, out var pipeServer))
|
{
|
throw new InvalidOperationException($"找不到客户端 {clientId}");
|
}
|
|
_lock.EnterWriteLock();
|
try
|
{
|
try
|
{
|
pipeServer.WriteMessage(message);
|
}
|
catch (Exception ex)
|
{
|
OnError?.Invoke(this, new PipeErrorEventArgs($"向客户端 {clientId} 发送消息时发生错误", ex));
|
}
|
}
|
finally
|
{
|
_lock.ExitWriteLock();
|
}
|
}
|
|
public void BroadcastMessage(string message)
|
{
|
_lock.EnterWriteLock();
|
try
|
{
|
foreach (var pipeServer in _activeConnections.Values)
|
{
|
pipeServer.WriteMessage(message);
|
}
|
}
|
finally
|
{
|
_lock.ExitWriteLock();
|
}
|
}
|
|
public void Stop()
|
{
|
_isRunning = false;
|
_cancellationTokenSource.Cancel();
|
}
|
|
public void Dispose()
|
{
|
if (!_isDisposed)
|
{
|
Stop();
|
foreach (var pipeServer in _activeConnections.Values)
|
{
|
pipeServer.Dispose();
|
}
|
_activeConnections.Clear();
|
_cancellationTokenSource.Dispose();
|
_lock.Dispose();
|
_isDisposed = true;
|
}
|
}
|
}
|
}
|
