using System; using System.Collections.Concurrent; using System.ComponentModel; using System.Net.Sockets; using System.Text; using System.Threading.Tasks; using Newtonsoft.Json; using TeamAAS.Communication.Attributes; using TeamAAS.Communication.Base; using TeamAAS.Communication.Interfaces; using TeamAAS.Communication.Enums; namespace TeamAAS.Communication.Devices { /// /// TCP 客户端(主动连接外部设备),纯透传。 /// [Communication("TCP 客户端", "基础通讯", "TCP 主动连接,纯透传")] public class TcpClientCommunication : BindableCommunicationBase { [JsonIgnore] private TcpClient _tcpClient; [JsonIgnore] private NetworkStream _stream; private readonly ConcurrentQueue _received = new ConcurrentQueue(); [JsonIgnore] private readonly StringBuilder _receiveBuffer = new StringBuilder(); [JsonIgnore] private System.Threading.Timer _reconnectTimer; [JsonIgnore] private volatile bool _hasConnectedOnce; [JsonIgnore] private volatile bool _manualDisconnect; [JsonIgnore] private volatile bool _isReconnecting; private string _ipAddress = "127.0.0.1"; [Category("I.客户端配置"), DisplayName("1.对端 IP 地址"), Description("对端 IP 地址")] public string IpAddress { get { return _ipAddress; } set { if (SetProperty(ref _ipAddress, value)) Notify(nameof(EndpointUrl)); } } private int _port = 5000; [Category("I.客户端配置"), DisplayName("2.对端端口"), Description("对端端口")] public int Port { get { return _port; } set { if (SetProperty(ref _port, value)) Notify(nameof(EndpointUrl)); } } private Terminator _terminator = Terminator.None; [Category("III.数据格式"), DisplayName("结束符"), Description("发送时自动附加、接收时自动去除的结束符")] public Terminator Terminator { get { return _terminator; } set { SetProperty(ref _terminator, value); } } private DataEncoding _dataEncoding = DataEncoding.Default; [Category("III.数据格式"), DisplayName("编码格式"), Description("收发数据的编码格式")] public DataEncoding DataEncoding { get { return _dataEncoding; } set { SetProperty(ref _dataEncoding, value); } } [Browsable(false)] public override string EndpointUrl { get { return $"tcp://{IpAddress}:{Port}"; } set { if (!string.IsNullOrWhiteSpace(value) && value.StartsWith("tcp://")) { var uri = new Uri(value); IpAddress = uri.Host; Port = uri.Port; } } } [JsonIgnore, Browsable(false)] public override bool IsConnected => (_tcpClient != null && _tcpClient.Connected); public override event Action ConnectChangedEvent; public override event Action DataReceivedEvent; public override void Connect() { if (string.IsNullOrWhiteSpace(IpAddress)) throw new InvalidOperationException("TCP 客户端 IP 不能为空。"); if (Port <= 0) throw new InvalidOperationException("TCP 客户端端口必须大于 0。"); Disconnect(); _tcpClient = new TcpClient(); _tcpClient.Connect(IpAddress, Port); _stream = _tcpClient.GetStream(); _manualDisconnect = false; _hasConnectedOnce = true; StopReconnectTimer(); _ = Task.Run(ReceiveLoop); ConnectChangedEvent?.Invoke(this, IsConnected); Notify(nameof(IsConnected)); } public override Task ConnectAsync() { return Task.Run(() => Connect()); } public override void Disconnect() { StopReconnectTimer(); _manualDisconnect = true; if (_stream != null) { _stream.Dispose(); _stream = null; } if (_tcpClient != null) { _tcpClient.Dispose(); _tcpClient = null; } ConnectChangedEvent?.Invoke(this, false); Notify(nameof(IsConnected)); } public override void Dispose() { Disconnect(); } private async Task ReceiveLoop() { var buffer = new byte[4096]; try { while (_stream != null && IsConnected) { int n = await _stream.ReadAsync(buffer, 0, buffer.Length); if (n <= 0) break; _receiveBuffer.Append(GetEncoding().GetString(buffer, 0, n)); FlushReceivedMessages(_receiveBuffer); } } catch { } finally { if (_hasConnectedOnce && !_manualDisconnect) StartAutoReconnect(); } } /// /// 获取编码 /// public Encoding GetEncoding() { switch (DataEncoding) { case DataEncoding.ASCII: return Encoding.ASCII; case DataEncoding.UTF7: return Encoding.UTF7; case DataEncoding.UTF8: return Encoding.UTF8; case DataEncoding.UTF32: return Encoding.UTF32; case DataEncoding.Unicode: return Encoding.Unicode; case DataEncoding.BigEndianUnicode: return Encoding.BigEndianUnicode; case DataEncoding.GB2312: return Encoding.GetEncoding("gb2312"); default: return Encoding.Default; } } /// /// 获取结束符字符串 /// public string GetTerminatorString() { switch (Terminator) { case Terminator.CR: return "\r"; case Terminator.LF: return "\n"; case Terminator.CRLF: return "\r\n"; case Terminator.None: default: return string.Empty; } } /// /// 去除数据尾部结束符 /// public string TrimTerminator(string text) { if (string.IsNullOrEmpty(text)) return text; var term = GetTerminatorString(); if (string.IsNullOrEmpty(term)) return text; return text.EndsWith(term) ? text.Substring(0, text.Length - term.Length) : text; } /// /// 按结束符分帧:收到完整结束符才触发接收事件;无结束符时直接触发 /// private void FlushReceivedMessages(StringBuilder receiveBuffer) { var term = GetTerminatorString(); if (string.IsNullOrEmpty(term)) { if (receiveBuffer.Length > 0) { var text = receiveBuffer.ToString(); receiveBuffer.Clear(); _received.Enqueue(text); DataReceivedEvent?.Invoke(this, text); } return; } string content = receiveBuffer.ToString(); int idx; while ((idx = content.IndexOf(term, StringComparison.Ordinal)) >= 0) { var msg = content.Substring(0, idx); content = content.Substring(idx + term.Length); _received.Enqueue(msg); DataReceivedEvent?.Invoke(this, msg); } receiveBuffer.Clear(); receiveBuffer.Append(content); } /// /// 自动重连:初始化连接成功过,掉线后每5秒尝试连接一次 /// private void StartAutoReconnect() { if (_reconnectTimer != null) return; if (_stream != null) { try { _stream.Dispose(); } catch { } _stream = null; } if (_tcpClient != null) { try { _tcpClient.Dispose(); } catch { } _tcpClient = null; } ConnectChangedEvent?.Invoke(this, false); Notify(nameof(IsConnected)); _reconnectTimer = new System.Threading.Timer(ReconnectTimer_Elapsed, null, TimeSpan.FromSeconds(5), TimeSpan.FromSeconds(5)); } private void ReconnectTimer_Elapsed(object state) { if (_manualDisconnect) { StopReconnectTimer(); return; } if (_isReconnecting) return; _isReconnecting = true; try { if (_stream != null) { try { _stream.Dispose(); } catch { } _stream = null; } if (_tcpClient != null) { try { _tcpClient.Dispose(); } catch { } _tcpClient = null; } var client = new TcpClient(); try { client.Connect(IpAddress, Port); _tcpClient = client; _stream = _tcpClient.GetStream(); StopReconnectTimer(); _ = Task.Run(ReceiveLoop); ConnectChangedEvent?.Invoke(this, true); Notify(nameof(IsConnected)); } catch { try { client.Dispose(); } catch { } } } finally { _isReconnecting = false; } } private void StopReconnectTimer() { var t = _reconnectTimer; _reconnectTimer = null; if (t != null) { try { t.Dispose(); } catch { } } } public void Send(string text) { if (_stream == null) throw new InvalidOperationException("TCP 客户端未连接。"); var data = GetEncoding().GetBytes((text ?? string.Empty) + GetTerminatorString()); _stream.Write(data, 0, data.Length); } public override object ReadValue(string address) { _received.TryDequeue(out var text); return text; } public override Task ReadValueAsync(string address) { return Task.FromResult(ReadValue(address)); } public override void WriteValue(string address, object value) { Send(value?.ToString() ?? string.Empty); } public override Task WriteValueAsync(string address, object value) { Send(value?.ToString() ?? string.Empty); return Task.CompletedTask; } } }