using System;
using System.IO;
using System.Net;
using System.Net.Sockets;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using TeamAAS.Robot.Enums;
using TeamAAS.Communication.Enums;
namespace TeamAAS.Robot.Core
{
///
/// 机器人 TCP 通讯封装类。
/// 支持 Client/Server 双模式、结束符分割、请求-响应、断线重连。
///
public class RobotTcpClient : IDisposable
{
private readonly string _ip;
private readonly int _port;
private readonly TCPConnectType _connectType;
private readonly Terminator _terminator;
private readonly DataEncoding _dataEncoding;
private TcpClient _tcpClient;
private TcpListener _tcpListener;
private TcpClient _serverSession;
private NetworkStream _serverStream;
private readonly AutoResetEvent _receiveEvent = new AutoResetEvent(false);
private byte[] _receivedData;
private readonly object _sendLock = new object();
private readonly object _sessionLock = new object();
private Task _reconnectTask;
private CancellationTokenSource _reconnectCts;
private bool _isDisposed;
private string _terminatorString;
private byte[] _terminatorBytes;
public bool IsConnected
{
get
{
if (_connectType == TCPConnectType.Client)
return _tcpClient?.Connected ?? false;
else
return _serverSession?.Connected ?? false;
}
}
public RobotTcpClient(
string ip,
int port,
TCPConnectType connectType,
Terminator terminator,
DataEncoding dataEncoding)
{
_ip = ip;
_port = port;
_connectType = connectType;
_terminator = terminator;
_dataEncoding = dataEncoding;
_terminatorString = terminator switch
{
Terminator.CR => "\r",
Terminator.LF => "\n",
Terminator.CRLF => "\r\n",
_ => "\r\n"
};
_terminatorBytes = GetEncoding().GetBytes(_terminatorString);
}
public void Connect()
{
if (_connectType == TCPConnectType.Client)
ConnectClient();
else
ConnectServer();
}
public async Task ConnectAsync()
{
if (_connectType == TCPConnectType.Client)
{
await ConnectClientAsync();
}
else
{
ConnectServer();
}
}
private void ConnectClient()
{
_tcpClient = new TcpClient();
_tcpClient.Connect(IPAddress.Parse(_ip), _port);
StartReconnectLoop();
StartReceiveLoop();
OnConnected?.Invoke(null);
}
private async Task ConnectClientAsync()
{
_tcpClient = new TcpClient();
await _tcpClient.ConnectAsync(IPAddress.Parse(_ip), _port);
StartReconnectLoop();
StartReceiveLoop();
OnConnected?.Invoke(null);
}
private void ConnectServer()
{
_tcpListener = new TcpListener(IPAddress.Any, _port);
_tcpListener.Start();
AcceptClientLoop();
}
private async void AcceptClientLoop()
{
while (!_isDisposed && _tcpListener != null)
{
try
{
var client = await _tcpListener.AcceptTcpClientAsync();
lock (_sessionLock)
{
_serverSession = client;
_serverStream = client.GetStream();
}
StartServerReceiveLoop(client);
OnConnected?.Invoke(null);
}
catch (ObjectDisposedException) { break; }
catch (Exception)
{
if (!_isDisposed)
await Task.Delay(1000);
}
}
}
private void StartReceiveLoop()
{
var stream = _tcpClient.GetStream();
Task.Run(() => ReceiveDataLoop(stream));
}
private void StartServerReceiveLoop(TcpClient client)
{
var stream = client.GetStream();
Task.Run(() => ReceiveDataLoop(stream));
}
private async void ReceiveDataLoop(NetworkStream stream)
{
byte[] buffer = new byte[4096];
MemoryStream ms = new MemoryStream();
while (!_isDisposed)
{
try
{
int bytesRead = await stream.ReadAsync(buffer, 0, buffer.Length);
if (bytesRead == 0) break;
ms.Write(buffer, 0, bytesRead);
while (ContainsTerminator(ms))
{
byte[] message = ExtractMessage(ms);
ProcessReceivedData(message);
}
}
catch (IOException)
{
break;
}
catch (Exception)
{
if (!_isDisposed)
await Task.Delay(100);
else
break;
}
}
if (_connectType == TCPConnectType.Client)
{
OnDisconnected?.Invoke(null);
}
else
{
lock (_sessionLock)
{
if (_serverSession != null && _serverSession == GetCurrentSession())
{
_serverSession = null;
_serverStream = null;
}
}
OnDisconnected?.Invoke(null);
}
}
private TcpClient GetCurrentSession()
{
return _serverSession;
}
private bool ContainsTerminator(MemoryStream ms)
{
if (ms.Length < _terminatorBytes.Length) return false;
byte[] data = ms.ToArray();
for (int i = 0; i <= data.Length - _terminatorBytes.Length; i++)
{
bool match = true;
for (int j = 0; j < _terminatorBytes.Length; j++)
{
if (data[i + j] != _terminatorBytes[j])
{
match = false;
break;
}
}
if (match) return true;
}
return false;
}
private byte[] ExtractMessage(MemoryStream ms)
{
byte[] data = ms.ToArray();
int endIndex = -1;
for (int i = 0; i <= data.Length - _terminatorBytes.Length; i++)
{
bool match = true;
for (int j = 0; j < _terminatorBytes.Length; j++)
{
if (data[i + j] != _terminatorBytes[j])
{
match = false;
break;
}
}
if (match)
{
endIndex = i + _terminatorBytes.Length;
break;
}
}
byte[] message;
if (endIndex > 0)
{
message = new byte[endIndex];
Array.Copy(data, message, endIndex);
ms.Position = 0;
ms.SetLength(0);
if (endIndex < data.Length)
{
ms.Write(data, endIndex, data.Length - endIndex);
}
}
else
{
message = data;
ms.Position = 0;
ms.SetLength(0);
}
return message;
}
private void ProcessReceivedData(byte[] data)
{
_receivedData = data;
OnReceived?.Invoke(data);
_receiveEvent.Set();
}
private void StartReconnectLoop()
{
_reconnectCts = new CancellationTokenSource();
_reconnectTask = Task.Run(async () =>
{
while (!_reconnectCts.Token.IsCancellationRequested)
{
await Task.Delay(1000, _reconnectCts.Token);
if (_reconnectCts.Token.IsCancellationRequested) break;
if (_connectType == TCPConnectType.Client && !IsConnected)
{
try
{
_tcpClient?.Close();
_tcpClient = new TcpClient();
_tcpClient.Connect(IPAddress.Parse(_ip), _port);
StartReceiveLoop();
}
catch { }
}
}
});
}
public void Disconnect()
{
_isDisposed = true;
_reconnectCts?.Cancel();
_tcpClient?.Close();
_tcpListener?.Stop();
_serverSession?.Close();
}
public string SendAndReceive(string message, int timeoutMs = 5000)
{
byte[] data = SendAndReceiveBytes(GetEncoding().GetBytes(message + _terminatorString), timeoutMs);
return GetEncoding().GetString(data).TrimEnd('\r', '\n');
}
public async Task SendAndReceiveAsync(string message, int timeoutMs = 5000)
{
byte[] data = await SendAndReceiveBytesAsync(GetEncoding().GetBytes(message + _terminatorString), timeoutMs);
return GetEncoding().GetString(data).TrimEnd('\r', '\n');
}
private byte[] SendAndReceiveBytes(byte[] data, int timeoutMs)
{
lock (_sendLock)
{
SendBytesInternal(data);
OnSent?.Invoke(GetEncoding().GetString(data));
if (!_receiveEvent.WaitOne(timeoutMs))
throw new TimeoutException($"等待机器人响应超时({timeoutMs}ms):已发送 \"{GetEncoding().GetString(data).Trim()}\"");
return _receivedData ?? Array.Empty();
}
}
private async Task SendAndReceiveBytesAsync(byte[] data, int timeoutMs)
{
await Task.Run(() =>
{
lock (_sendLock)
{
SendBytesInternal(data);
OnSent?.Invoke(GetEncoding().GetString(data));
if (!_receiveEvent.WaitOne(timeoutMs))
throw new TimeoutException($"等待机器人响应超时({timeoutMs}ms):已发送 \"{GetEncoding().GetString(data).Trim()}\"");
}
});
return _receivedData ?? Array.Empty();
}
public void Send(string message)
{
byte[] data = GetEncoding().GetBytes(message + _terminatorString);
SendBytesInternal(data);
OnSent?.Invoke(message);
}
public async Task SendAsync(string message)
{
byte[] data = GetEncoding().GetBytes(message + _terminatorString);
await SendBytesInternalAsync(data);
OnSent?.Invoke(message);
}
private void SendBytesInternal(byte[] data)
{
if (_connectType == TCPConnectType.Client)
{
if (_tcpClient != null && _tcpClient.Connected)
{
var stream = _tcpClient.GetStream();
stream.Write(data, 0, data.Length);
stream.Flush();
}
}
else
{
lock (_sessionLock)
{
if (_serverStream != null && _serverSession != null && _serverSession.Connected)
{
_serverStream.Write(data, 0, data.Length);
_serverStream.Flush();
}
}
}
}
private async Task SendBytesInternalAsync(byte[] data)
{
if (_connectType == TCPConnectType.Client)
{
if (_tcpClient != null && _tcpClient.Connected)
{
var stream = _tcpClient.GetStream();
await stream.WriteAsync(data, 0, data.Length);
await stream.FlushAsync();
}
}
else
{
lock (_sessionLock)
{
if (_serverStream != null && _serverSession != null && _serverSession.Connected)
{
_serverStream.Write(data, 0, data.Length);
_serverStream.Flush();
}
}
}
}
public Encoding GetEncoding()
{
return _dataEncoding switch
{
DataEncoding.ASCII => Encoding.ASCII,
DataEncoding.UTF7 => Encoding.UTF7,
DataEncoding.UTF8 => Encoding.UTF8,
DataEncoding.UTF32 => Encoding.UTF32,
DataEncoding.Unicode => Encoding.Unicode,
_ => Encoding.Default
};
}
public void Dispose()
{
Disconnect();
_tcpClient?.Dispose();
_serverSession?.Dispose();
_serverStream?.Dispose();
_receiveEvent.Dispose();
_reconnectCts?.Dispose();
}
public event Action OnConnected;
public event Action OnDisconnected;
public event Action OnReceived;
public event Action OnSent;
}
}