Use DotNetty for better TCP and UDP experience

Modbus.Net/Modbus.Net.Modbus.NA200H/Modbus.Net.Modbus.NA200H.csproj

@@ -5,7 +5,7 @@
     <AssemblyName>Modbus.Net.Modbus.NA200H</AssemblyName>
     <RootNamespace>Modbus.Net.Modbus.NA200H</RootNamespace>
     <PackageId>Modbus.Net.Modbus.NA200H</PackageId>
-    <Version>1.4.0-beta04</Version>
+    <Version>1.4.1-beta05</Version>
     <Authors>Chris L.(Luo Sheng)</Authors>
     <Company>Hangzhou Delian Science Technology Co.,Ltd.</Company>
     <Product>Modbus.Net.Modbus</Product>

Modbus.Net/Modbus.Net.Modbus/Modbus.Net.Modbus.csproj

@@ -5,7 +5,7 @@
     <AssemblyName>Modbus.Net.Modbus</AssemblyName>
     <RootNamespace>Modbus.Net.Modbus</RootNamespace>
     <PackageId>Modbus.Net.Modbus</PackageId>
-    <Version>1.4.0-beta04</Version>
+    <Version>1.4.1-beta05</Version>
     <Authors>Chris L.(Luo Sheng)</Authors>
     <Company>Hangzhou Delian Science Technology Co.,Ltd.</Company>
     <Product>Modbus.Net.Modbus</Product>

Modbus.Net/Modbus.Net.Modbus/ModbusAsciiInTcpProtocolLinker.cs

@@ -24,7 +24,7 @@ namespace Modbus.Net.Modbus
         public ModbusAsciiInTcpProtocolLinker(string ip, int port)
             : base(ip, port)
         {
-            ((BaseConnector)BaseConnector).AddController(new FifoController(int.Parse(ConfigurationReader.GetValue("TCP:" + ip + ":" + port, "FetchSleepTime")), lengthCalc: content => { if (content[0] != 0x3a) return 0; for (int i = 1; i < content.Length; i++) { if (content[i - 1] == 0x0D && content[i] == 0x0A) return i + 1; } return -1; }, waitingListMaxCount: ConfigurationReader.GetValue("TCP:" + ip + ":" + port, "WaitingListCount") != null ? int.Parse(ConfigurationReader.GetValue("TCP:" + ip + ":" + port, "WaitingListCount")) : null));
+            ((EventHandlerConnector)BaseConnector).AddController(new FifoController(int.Parse(ConfigurationReader.GetValue("TCP:" + ip + ":" + port, "FetchSleepTime")), lengthCalc: content => { if (content[0] != 0x3a) return 0; for (int i = 1; i < content.Length; i++) { if (content[i - 1] == 0x0D && content[i] == 0x0A) return i + 1; } return -1; }, waitingListMaxCount: ConfigurationReader.GetValue("TCP:" + ip + ":" + port, "WaitingListCount") != null ? int.Parse(ConfigurationReader.GetValue("TCP:" + ip + ":" + port, "WaitingListCount")) : null));
         }
 
         /// <summary>

Modbus.Net/Modbus.Net.Modbus/ModbusAsciiInUdpProtocolLinker.cs

@@ -24,7 +24,7 @@ namespace Modbus.Net.Modbus
         public ModbusAsciiInUdpProtocolLinker(string ip, int port)
             : base(ip, port)
         {
-            ((BaseConnector)BaseConnector).AddController(new FifoController(int.Parse(ConfigurationReader.GetValue("UDP:" + ip + ":" + port, "FetchSleepTime")), lengthCalc: content => { if (content[0] != 0x3a) return 0; for (int i = 1; i < content.Length; i++) { if (content[i - 1] == 0x0D && content[i] == 0x0A) return i + 1; } return -1; }, waitingListMaxCount: ConfigurationReader.GetValue("UDP:" + ip + ":" + port, "WaitingListCount") != null ? int.Parse(ConfigurationReader.GetValue("UDP:" + ip + ":" + port, "WaitingListCount")) : null));
+            ((EventHandlerConnector)BaseConnector).AddController(new FifoController(int.Parse(ConfigurationReader.GetValue("UDP:" + ip + ":" + port, "FetchSleepTime")), lengthCalc: content => { if (content[0] != 0x3a) return 0; for (int i = 1; i < content.Length; i++) { if (content[i - 1] == 0x0D && content[i] == 0x0A) return i + 1; } return -1; }, waitingListMaxCount: ConfigurationReader.GetValue("UDP:" + ip + ":" + port, "WaitingListCount") != null ? int.Parse(ConfigurationReader.GetValue("UDP:" + ip + ":" + port, "WaitingListCount")) : null));
         }
 
         /// <summary>

Modbus.Net/Modbus.Net.Modbus/ModbusRtuInTcpProtocolLinker.cs

@@ -24,7 +24,7 @@ namespace Modbus.Net.Modbus
         public ModbusRtuInTcpProtocolLinker(string ip, int port)
             : base(ip, port)
         {
-            ((BaseConnector)BaseConnector).AddController(new FifoController(int.Parse(ConfigurationReader.GetValue("TCP:" + ip + ":" + port, "FetchSleepTime")), lengthCalc: content => { if (content[1] == 5 || content[1] == 6 || content[1] == 15 || content[1] == 16 || content[1] == 21) return 8; else return DuplicateWithCount.GetDuplcateFunc(new List<int> { 2 }, 5).Invoke(content); }, waitingListMaxCount: ConfigurationReader.GetValue("TCP:" + ip + ":" + port, "WaitingListCount") != null ? int.Parse(ConfigurationReader.GetValue("TCP:" + ip + ":" + port, "WaitingListCount")) : null));
+            ((EventHandlerConnector)BaseConnector).AddController(new FifoController(int.Parse(ConfigurationReader.GetValue("TCP:" + ip + ":" + port, "FetchSleepTime")), lengthCalc: content => { if (content[1] == 5 || content[1] == 6 || content[1] == 15 || content[1] == 16 || content[1] == 21) return 8; else return DuplicateWithCount.GetDuplcateFunc(new List<int> { 2 }, 5).Invoke(content); }, waitingListMaxCount: ConfigurationReader.GetValue("TCP:" + ip + ":" + port, "WaitingListCount") != null ? int.Parse(ConfigurationReader.GetValue("TCP:" + ip + ":" + port, "WaitingListCount")) : null));
         }
 
         /// <summary>

Modbus.Net/Modbus.Net.Modbus/ModbusRtuInUdpProtocolLinker.cs

@@ -24,7 +24,7 @@ namespace Modbus.Net.Modbus
         public ModbusRtuInUdpProtocolLinker(string ip, int port)
             : base(ip, port)
         {
-            ((BaseConnector)BaseConnector).AddController(new FifoController(int.Parse(ConfigurationReader.GetValue("UDP:" + ip + ":" + port, "FetchSleepTime")), lengthCalc: content => { if (content[1] == 5 || content[1] == 6 || content[1] == 15 || content[1] == 16 || content[1] == 21) return 8; else return DuplicateWithCount.GetDuplcateFunc(new List<int> { 2 }, 5).Invoke(content); }, waitingListMaxCount: ConfigurationReader.GetValue("UDP:" + ip + ":" + port, "WaitingListCount") != null ? int.Parse(ConfigurationReader.GetValue("UDP:" + ip + ":" + port, "WaitingListCount")) : null));
+            ((EventHandlerConnector)BaseConnector).AddController(new FifoController(int.Parse(ConfigurationReader.GetValue("UDP:" + ip + ":" + port, "FetchSleepTime")), lengthCalc: content => { if (content[1] == 5 || content[1] == 6 || content[1] == 15 || content[1] == 16 || content[1] == 21) return 8; else return DuplicateWithCount.GetDuplcateFunc(new List<int> { 2 }, 5).Invoke(content); }, waitingListMaxCount: ConfigurationReader.GetValue("UDP:" + ip + ":" + port, "WaitingListCount") != null ? int.Parse(ConfigurationReader.GetValue("UDP:" + ip + ":" + port, "WaitingListCount")) : null));
         }
 
         /// <summary>

Modbus.Net/Modbus.Net.Modbus/ModbusTcpProtocolLinker.cs

@@ -23,7 +23,7 @@ namespace Modbus.Net.Modbus
         /// <param name="port">端口</param>
         public ModbusTcpProtocolLinker(string ip, int port) : base(ip, port)
         {
-            ((BaseConnector)BaseConnector).AddController(new FifoController(int.Parse(ConfigurationReader.GetValue("TCP:" + ip + ":" + port, "FetchSleepTime")), lengthCalc: DuplicateWithCount.GetDuplcateFunc(new List<int> { 4, 5 }, 6), waitingListMaxCount: ConfigurationReader.GetValue("TCP:" + ip + ":" + port, "WaitingListCount") != null ? int.Parse(ConfigurationReader.GetValue("TCP:" + ip + ":" + port, "WaitingListCount")) : null));
+            ((EventHandlerConnector)BaseConnector).AddController(new FifoController(int.Parse(ConfigurationReader.GetValue("TCP:" + ip + ":" + port, "FetchSleepTime")), lengthCalc: DuplicateWithCount.GetDuplcateFunc(new List<int> { 4, 5 }, 6), waitingListMaxCount: ConfigurationReader.GetValue("TCP:" + ip + ":" + port, "WaitingListCount") != null ? int.Parse(ConfigurationReader.GetValue("TCP:" + ip + ":" + port, "WaitingListCount")) : null));
         }
 
         /// <summary>

Modbus.Net/Modbus.Net.Modbus/ModbusUdpProtocolLinker.cs

@@ -12,7 +12,7 @@ namespace Modbus.Net.Modbus
         /// </summary>
         /// <param name="ip">IP地址</param>
         public ModbusUdpProtocolLinker(string ip)
-            : this(ip, int.Parse(ConfigurationReader.GetValue("UDP:" + ip, "ModbusPort")))
+            : this(ip, int.Parse(ConfigurationReader.GetValueDirect("UDP:" + ip, "ModbusPort") ?? ConfigurationReader.GetValueDirect("UDP:Modbus", "ModbusPort")))
         {
         }
 
@@ -23,7 +23,7 @@ namespace Modbus.Net.Modbus
         /// <param name="port">端口</param>
         public ModbusUdpProtocolLinker(string ip, int port) : base(ip, port)
         {
-            ((BaseConnector)BaseConnector).AddController(new FifoController(int.Parse(ConfigurationReader.GetValue("UDP:" + ip + ":" + port, "FetchSleepTime")), lengthCalc: DuplicateWithCount.GetDuplcateFunc(new List<int> { 4, 5 }, 6), waitingListMaxCount: ConfigurationReader.GetValue("UDP:" + ip + ":" + port, "WaitingListCount") != null ? int.Parse(ConfigurationReader.GetValue("UDP:" + ip + ":" + port, "WaitingListCount")) : null));
+            ((EventHandlerConnector)BaseConnector).AddController(new FifoController(int.Parse(ConfigurationReader.GetValue("UDP:" + ip + ":" + port, "FetchSleepTime")), lengthCalc: DuplicateWithCount.GetDuplcateFunc(new List<int> { 4, 5 }, 6), waitingListMaxCount: ConfigurationReader.GetValue("UDP:" + ip + ":" + port, "WaitingListCount") != null ? int.Parse(ConfigurationReader.GetValue("UDP:" + ip + ":" + port, "WaitingListCount")) : null));
         }
 
         /// <summary>

Modbus.Net/Modbus.Net.OPC/Modbus.Net.OPC.csproj

@@ -5,7 +5,7 @@
     <AssemblyName>Modbus.Net.OPC</AssemblyName>
     <RootNamespace>Modbus.Net.OPC</RootNamespace>
     <PackageId>Modbus.Net.OPC</PackageId>
-    <Version>1.4.0-beta04</Version>
+    <Version>1.4.1-beta05</Version>
     <Authors>Chris L.(Luo Sheng)</Authors>
     <Company>Hangzhou Delian Science Technology Co.,Ltd.</Company>
     <Product>Modbus.Net.OPC</Product>

Modbus.Net/Modbus.Net.Siemens/Modbus.Net.Siemens.csproj

@@ -5,7 +5,7 @@
     <AssemblyName>Modbus.Net.Siemens</AssemblyName>
     <RootNamespace>Modbus.Net.Siemens</RootNamespace>
     <PackageId>Modbus.Net.Siemens</PackageId>
-    <Version>1.4.0-beta04</Version>
+    <Version>1.4.1-beta05</Version>
     <Authors>Chris L.(Luo Sheng)</Authors>
     <Company>Hangzhou Delian Science Technology Co.,Ltd.</Company>
     <Description>Modbus.Net Siemens Profinet Implementation</Description>

Modbus.Net/Modbus.Net.Siemens/SiemensTcpProtocolLinker.cs

@@ -25,7 +25,7 @@ namespace Modbus.Net.Siemens
         public SiemensTcpProtocolLinker(string ip, int port)
             : base(ip, port)
         {
-            ((BaseConnector)BaseConnector).AddController(new MatchDirectlySendController(new ICollection<(int, int)>[] { new List<(int, int)> { (11, 11), (12, 12) } }, DuplicateWithCount.GetDuplcateFunc(new List<int> { 2, 3 }, 0), waitingListMaxCount: ConfigurationReader.GetValue("TCP:" + ip + ":" + port, "WaitingListCount") != null ? int.Parse(ConfigurationReader.GetValue("TCP:" + ip + ":" + port, "WaitingListCount")) : null));
+            ((EventHandlerConnector)BaseConnector).AddController(new MatchDirectlySendController(new ICollection<(int, int)>[] { new List<(int, int)> { (11, 11), (12, 12) } }, DuplicateWithCount.GetDuplcateFunc(new List<int> { 2, 3 }, 0), waitingListMaxCount: ConfigurationReader.GetValue("TCP:" + ip + ":" + port, "WaitingListCount") != null ? int.Parse(ConfigurationReader.GetValue("TCP:" + ip + ":" + port, "WaitingListCount")) : null));
         }
 
         /// <summary>

Modbus.Net/Modbus.Net/Connector/EventHandlerConnector.cs

@@ -0,0 +1,172 @@
+using DotNetty.Transport.Channels;
+using Microsoft.Extensions.Logging;
+using Nito.AsyncEx;
+using System;
+using System.Threading.Tasks;
+
+namespace Modbus.Net
+{
+    /// <inheritdoc />
+    public abstract class EventHandlerConnector : EventHandlerConnector<byte[], byte[]>
+    {
+        /// <inheridoc />
+        public override bool IsSharable => true;
+
+        private static readonly ILogger<EventHandlerConnector> logger = LogProvider.CreateLogger<EventHandlerConnector>();
+
+        /// <summary>
+        ///     发送锁
+        /// </summary>
+        protected abstract AsyncLock Lock { get; }
+
+        /// <summary>
+        ///     是否为全双工
+        /// </summary>
+        public bool IsFullDuplex { get; }
+
+        /// <summary>
+        ///     发送超时时间
+        /// </summary>
+        protected abstract int TimeoutTime { get; set; }
+
+        /// <summary>
+        ///     构造器
+        /// </summary>
+        /// <param name="timeoutTime">发送超时时间</param>
+        /// <param name="isFullDuplex">是否为全双工</param>
+        protected EventHandlerConnector(int timeoutTime = 10000, bool isFullDuplex = true)
+        {
+            IsFullDuplex = isFullDuplex;
+            if (timeoutTime < -1) timeoutTime = -1;
+            TimeoutTime = timeoutTime;
+        }
+
+        /// <inheritdoc />
+        public override async Task<byte[]> SendMsgAsync(byte[] message)
+        {
+            var ans = await SendMsgInner(message);
+            if (ans == null) return new byte[0];
+            return ans.ReceiveMessage;
+        }
+
+        /// <summary>
+        ///     发送内部
+        /// </summary>
+        /// <param name="message">发送的信息</param>
+        /// <returns>发送信息的定义</returns>
+        protected async Task<MessageWaitingDef> SendMsgInner(byte[] message)
+        {
+            IDisposable asyncLock = null;
+            try
+            {
+                var messageSendingdef = Controller.AddMessage(message);
+                if (messageSendingdef != null)
+                {
+                    if (!IsFullDuplex)
+                    {
+                        asyncLock = await Lock.LockAsync();
+                    }
+                    var success = messageSendingdef.SendMutex.WaitOne(TimeoutTime);
+                    if (success)
+                    {
+                        await SendMsgWithoutConfirm(message);
+                        success = messageSendingdef.ReceiveMutex.WaitOne(TimeoutTime);
+                        if (success)
+                        {
+                            return messageSendingdef;
+                        }
+                    }
+                    Controller.ForceRemoveWaitingMessage(messageSendingdef);
+                }
+                logger.LogInformation("Message is waiting in {0}. Cancel!", ConnectionToken);
+                return null;
+            }
+            catch (Exception e)
+            {
+                logger.LogError(e, "Connector {0} Send Error.", ConnectionToken);
+                return null;
+            }
+            finally
+            {
+                asyncLock?.Dispose();
+            }
+        }
+
+        /// <inheridoc />
+        public override void ChannelReadComplete(IChannelHandlerContext ctx)
+        {
+            ctx.Flush();
+        }
+
+        /// <inheridoc />
+        public override void ExceptionCaught(IChannelHandlerContext ctx, Exception e)
+        {
+            logger.LogError(e, e.ToString());
+            ctx.CloseAsync();
+        }
+    }
+
+    /// <summary>
+    ///     基础的协议连接类
+    /// </summary>
+    public abstract class EventHandlerConnector<TParamIn, TParamOut> : ChannelHandlerAdapter, IConnector<TParamIn, TParamOut> where TParamIn : class
+    {
+        /// <summary>
+        ///     数据返回代理参数
+        /// </summary>
+        /// <param name="sender"></param>
+        /// <param name="args"></param>
+        /// <returns></returns>
+        public delegate MessageReturnCallbackArgs<TParamIn> MessageReturnDelegate(object sender, MessageReturnArgs<TParamOut> args);
+
+        /// <summary>
+        ///     数据返回代理
+        /// </summary>
+        public event MessageReturnDelegate MessageReturn;
+
+        /// <summary>
+        ///     增加传输控制器
+        /// </summary>
+        /// <param name="controller">传输控制器</param>
+        public void AddController(IController controller)
+        {
+            Controller = controller;
+        }
+
+        /// <summary>
+        ///     传输控制器
+        /// </summary>
+        protected virtual IController Controller { get; set; }
+
+        /// <inheritdoc />
+        public abstract string ConnectionToken { get; }
+
+        /// <inheritdoc />
+        public abstract bool IsConnected { get; }
+
+        /// <inheritdoc />
+        public abstract Task<bool> ConnectAsync();
+
+        /// <inheritdoc />
+        public abstract bool Disconnect();
+
+        /// <inheritdoc />
+        public abstract Task<TParamOut> SendMsgAsync(TParamIn message);
+
+        /// <summary>
+        ///     发送数据，不确认
+        /// </summary>
+        /// <param name="message">需要发送的数据</param>
+        protected abstract Task SendMsgWithoutConfirm(TParamIn message);
+
+        /// <summary>
+        ///     数据返回代理函数
+        /// </summary>
+        /// <param name="receiveMessage"></param>
+        /// <returns></returns>
+        protected TParamIn InvokeReturnMessage(TParamOut receiveMessage)
+        {
+            return MessageReturn?.Invoke(this, new MessageReturnArgs<TParamOut> { ReturnMessage = receiveMessage })?.SendMessage;
+        }
+    }
+}

Modbus.Net/Modbus.Net/Connector/TcpConnector.cs

@@ -1,9 +1,13 @@
-using Microsoft.Extensions.Logging;
+using DotNetty.Buffers;
+using DotNetty.Common.Utilities;
+using DotNetty.Transport.Bootstrapping;
+using DotNetty.Transport.Channels;
+using DotNetty.Transport.Channels.Sockets;
+using Microsoft.Extensions.Logging;
 using Nito.AsyncEx;
 using System;
 using System.Linq;
-using System.Net.Sockets;
+using System.Net;
-using System.Threading;
 using System.Threading.Tasks;
 
 namespace Modbus.Net
@@ -12,29 +16,19 @@ namespace Modbus.Net
     ///     Socket收发类
     ///     作者：本类来源于CSDN，并由罗圣（Chris L.）根据实际需要修改
     /// </summary>
-    public class TcpConnector : BaseConnector, IDisposable
+    public class TcpConnector : EventHandlerConnector, IDisposable
     {
         private static readonly ILogger<TcpConnector> logger = LogProvider.CreateLogger<TcpConnector>();
 
         private readonly string _host;
         private readonly int _port;
 
-        /// <summary>
-        ///     1MB 的接收缓冲区
-        /// </summary>
-        private readonly byte[] _receiveBuffer = new byte[1024];
-
         private int _errorCount;
         private int _receiveCount;
 
         private int _sendCount;
 
-        private TcpClient _socketClient;
+        private IChannel Channel { get; set; }
-
-        private int _timeoutTime;
-
-        private Task _receiveThread;
-        private bool _taskCancel = false;
 
         /// <summary>
         ///     构造器
@@ -53,20 +47,10 @@ namespace Modbus.Net
         public override string ConnectionToken => _host;
 
         /// <inheritdoc />
-        protected override int TimeoutTime
+        protected override int TimeoutTime { get; set; }
-        {
-            get =>
-            _timeoutTime;
-            set
-            {
-                _timeoutTime = value;
-                if (_socketClient != null)
-                    _socketClient.ReceiveTimeout = _timeoutTime;
-            }
-        }
 
         /// <inheritdoc />
-        public override bool IsConnected => _socketClient?.Client != null && _socketClient.Connected;
+        public override bool IsConnected => Channel?.Open == true;
 
         /// <inheritdoc />
         protected override AsyncLock Lock { get; } = new AsyncLock();
@@ -92,10 +76,10 @@ namespace Modbus.Net
                 // Release managed resources
             }
             // Release unmanaged resources
-            if (_socketClient != null)
+            if (Channel != null)
             {
-                CloseClientSocket();
+                CloseClientSocket().Wait();
-                _socketClient = null;
+                Channel = null;
                 logger.LogDebug("Tcp client {ConnectionToken} Disposed", ConnectionToken);
             }
         }
@@ -114,28 +98,31 @@ namespace Modbus.Net
         {
             using (await Lock.LockAsync())
             {
-                if (_socketClient != null)
+                if (Channel != null)
                 {
-                    if (_socketClient.Connected)
+                    if (Channel.Open)
                         return true;
                 }
                 try
                 {
-                    _socketClient = new TcpClient
+                    var bootstrap = new Bootstrap();
-                    {
+                    bootstrap
-                        SendTimeout = TimeoutTime,
+                        .Group(new MultithreadEventLoopGroup())
-                        ReceiveTimeout = TimeoutTime
+                        .Channel<TcpSocketChannel>()
-                    };
+                        .Option(ChannelOption.TcpNodelay, true)
+                        .Option(ChannelOption.ConnectTimeout, TimeSpan.FromMilliseconds(TimeoutTime))
+                        .Handler(new ActionChannelInitializer<ISocketChannel>(channel =>
+                        {
+                            IChannelPipeline pipeline = channel.Pipeline;
 
-                    var cts = new CancellationTokenSource();
+                            pipeline.AddLast("handler", this);
-                    cts.CancelAfter(TimeoutTime);
+                        }));
-                    await _socketClient.ConnectAsync(_host, _port).WithCancellation(cts.Token);
 
-                    if (_socketClient.Connected)
+                    Channel = await bootstrap.ConnectAsync(new IPEndPoint(IPAddress.Parse(_host), _port));
+
+                    if (Channel.Open)
                     {
-                        _taskCancel = false;
                         Controller.SendStart();
-                        ReceiveMsgThreadStart();
                         logger.LogInformation("Tcp client {ConnectionToken} connected", ConnectionToken);
                         return true;
                     }
@@ -146,6 +133,9 @@ namespace Modbus.Net
                 catch (Exception err)
                 {
                     logger.LogError(err, "Tcp client {ConnectionToken} connect exception", ConnectionToken);
+
+                    RefreshErrorCount();
+
                     Dispose();
                     return false;
                 }
@@ -155,7 +145,7 @@ namespace Modbus.Net
         /// <inheritdoc />
         public override bool Disconnect()
         {
-            if (_socketClient == null)
+            if (Channel.Open)
                 return true;
 
             try
@@ -167,11 +157,14 @@ namespace Modbus.Net
             catch (Exception err)
             {
                 logger.LogError(err, "Tcp client {ConnectionToken} disconnected exception", ConnectionToken);
+
+                RefreshErrorCount();
+
                 return false;
             }
             finally
             {
-                _socketClient = null;
+                Channel = null;
             }
         }
 
@@ -185,75 +178,54 @@ namespace Modbus.Net
                 if (!IsConnected)
                     await ConnectAsync();
 
-                var stream = _socketClient.GetStream();
-
                 RefreshSendCount();
 
                 logger.LogDebug("Tcp client {ConnectionToken} send text len = {Length}", ConnectionToken, datagram.Length);
-                logger.LogDebug($"Tcp client {ConnectionToken} send: {String.Concat(datagram.Select(p => " " + p.ToString("X2")))}");
+                logger.LogDebug($"Tcp client {ConnectionToken} send: {string.Concat(datagram.Select(p => " " + p.ToString("X2")))}");
-                await stream.WriteAsync(datagram, 0, datagram.Length);
+                IByteBuffer buffer = Unpooled.Buffer();
+                buffer.WriteBytes(datagram);
+                await Channel.WriteAndFlushAsync(buffer);
             }
             catch (Exception err)
             {
                 logger.LogError(err, "Tcp client {ConnectionToken} send exception", ConnectionToken);
+
+                RefreshErrorCount();
+
                 Dispose();
             }
         }
 
-        /// <inheritdoc />
+        /// <inheridoc />
-        protected override void ReceiveMsgThreadStart()
+        public override async void ChannelRead(IChannelHandlerContext context, object message)
-        {
-            _receiveThread = Task.Run(ReceiveMessage);
-        }
-
-        /// <inheritdoc />
-        protected override void ReceiveMsgThreadStop()
-        {
-            _taskCancel = true;
-        }
-
-        /// <summary>
-        ///     接收返回消息
-        /// </summary>
-        /// <returns>返回的消息</returns>
-        protected async Task ReceiveMessage()
         {
             try
             {
-                while (!_taskCancel)
+                if (message is IByteBuffer buffer)
                 {
-                    if (_socketClient == null) break;
+                    byte[] msg = buffer.Array.Slice(buffer.ArrayOffset, buffer.ReadableBytes);
-                    NetworkStream stream = _socketClient.GetStream();
+                    logger.LogDebug("Tcp client {ConnectionToken} receive text len = {Length}", ConnectionToken,
-                    var len = await stream.ReadAsync(_receiveBuffer, 0, _receiveBuffer.Length);
+                       msg.Length);
-                    stream.Flush();
+                    logger.LogDebug(
-
+                        $"Tcp client {ConnectionToken} receive: {string.Concat(msg.Select(p => " " + p.ToString("X2")))}");
-                    // 异步接收回答
+                    var isMessageConfirmed = Controller.ConfirmMessage(msg);
-                    if (len > 0)
+                    if (isMessageConfirmed != null)
                     {
-                        byte[] receiveBytes = CheckReplyDatagram(len);
+                        foreach (var confirmed in isMessageConfirmed)
-                        logger.LogDebug("Tcp client {ConnectionToken} receive text len = {Length}", ConnectionToken,
-                            receiveBytes.Length);
-                        logger.LogDebug(
-                            $"Tcp client {ConnectionToken} receive: {String.Concat(receiveBytes.Select(p => " " + p.ToString("X2")))}");
-                        var isMessageConfirmed = Controller.ConfirmMessage(receiveBytes);
-                        if (isMessageConfirmed != null)
                         {
-                            foreach (var confirmed in isMessageConfirmed)
+                            if (confirmed.Item2 == false)
                             {
-                                if (confirmed.Item2 == false)
+                                var sendMessage = InvokeReturnMessage(confirmed.Item1);
+                                //主动传输事件
+                                if (sendMessage != null)
                                 {
-                                    var sendMessage = InvokeReturnMessage(confirmed.Item1);
+                                    await SendMsgWithoutConfirm(sendMessage);
-                                    //主动传输事件
-                                    if (sendMessage != null)
-                                    {
-                                        await SendMsgWithoutConfirm(sendMessage);
-                                    }
                                 }
                             }
                         }
-
-                        RefreshReceiveCount();
                     }
+
+                    RefreshReceiveCount();
                 }
             }
             catch (ObjectDisposedException)
@@ -263,23 +235,11 @@ namespace Modbus.Net
             catch (Exception err)
             {
                 logger.LogError(err, "Tcp client {ConnectionToken} receive exception", ConnectionToken);
-                //CloseClientSocket();
-            }
-        }
 
-        /// <summary>
-        ///     接收消息，并转换成字符串
-        /// </summary>
-        /// <param name="len">消息的长度</param>
-        private byte[] CheckReplyDatagram(int len)
-        {
-            var replyMessage = new byte[len];
-            Array.Copy(_receiveBuffer, replyMessage, len);
-
-            if (len <= 0)
                 RefreshErrorCount();
 
-            return replyMessage;
+                await CloseClientSocket();
+            }
         }
 
         private void RefreshSendCount()
@@ -300,24 +260,25 @@ namespace Modbus.Net
             logger.LogDebug("Tcp client {ConnectionToken} error count: {ErrorCount}", ConnectionToken, _errorCount);
         }
 
-        private void CloseClientSocket()
+        private async Task CloseClientSocket()
         {
             try
             {
                 Controller.SendStop();
                 Controller.Clear();
-                ReceiveMsgThreadStop();
+                if (Channel != null)
-                if (_socketClient != null)
                 {
-                    if (_socketClient.Connected)
+                    if (Channel.Open)
                     {
-                        _socketClient.Close();
+                        await Channel.CloseAsync();
                     }
                 }
             }
             catch (Exception ex)
             {
                 logger.LogError(ex, "Tcp client {ConnectionToken} client close exception", ConnectionToken);
+
+                RefreshErrorCount();
             }
         }
     }

Modbus.Net/Modbus.Net/Connector/UdpConnector.cs

@@ -1,9 +1,14 @@
-using Microsoft.Extensions.Logging;
+using DotNetty.Buffers;
+using DotNetty.Common.Utilities;
+using DotNetty.Transport.Bootstrapping;
+using DotNetty.Transport.Channels;
+using DotNetty.Transport.Channels.Sockets;
+using Microsoft.Extensions.Logging;
 using Nito.AsyncEx;
 using System;
 using System.Linq;
+using System.Net;
 using System.Net.Sockets;
-using System.Threading;
 using System.Threading.Tasks;
 
 namespace Modbus.Net
@@ -11,27 +16,19 @@ namespace Modbus.Net
     /// <summary>
     ///     Udp收发类
     /// </summary>
-    public class UdpConnector : BaseConnector, IDisposable
+    public class UdpConnector : EventHandlerConnector, IDisposable
     {
         private static readonly ILogger<UdpConnector> logger = LogProvider.CreateLogger<UdpConnector>();
 
         private readonly string _host;
         private readonly int _port;
 
-        /// <summary>
-        ///     1MB 的接收缓冲区
-        /// </summary>
-        private readonly byte[] _receiveBuffer = new byte[1024];
-
         private int _errorCount;
         private int _receiveCount;
 
         private int _sendCount;
 
-        private UdpClient _socketClient;
+        private IChannel Channel { get; set; }
-
-        private Task _receiveThread;
-        private bool _taskCancel = false;
 
         /// <summary>
         ///     构造器
@@ -53,7 +50,7 @@ namespace Modbus.Net
         protected override int TimeoutTime { get; set; }
 
         /// <inheritdoc />
-        public override bool IsConnected => _socketClient?.Client != null && _socketClient.Client.Connected;
+        public override bool IsConnected => Channel != null && Channel.Active;
 
         /// <inheritdoc />
         protected override AsyncLock Lock { get; } = new AsyncLock();
@@ -79,10 +76,10 @@ namespace Modbus.Net
                 // Release managed resources
             }
             // Release unmanaged resources
-            if (_socketClient != null)
+            if (Channel != null)
             {
-                CloseClientSocket();
+                CloseClientSocket().Wait();
-                _socketClient = null;
+                Channel = null;
                 logger.LogDebug("Udp client {ConnectionToken} Disposed", ConnectionToken);
             }
         }
@@ -101,23 +98,30 @@ namespace Modbus.Net
         {
             using (await Lock.LockAsync())
             {
-                if (_socketClient != null)
+                if (Channel != null)
                 {
                     return true;
                 }
                 try
                 {
-                    _socketClient = new UdpClient();
+                    var bootstrap = new Bootstrap();
+                    bootstrap
+                        .Group(new MultithreadEventLoopGroup())
+                        .Channel<SocketDatagramChannel>()
+                        .Option(ChannelOption.SoBroadcast, true)
+                        .Option(ChannelOption.ConnectTimeout, TimeSpan.FromMilliseconds(TimeoutTime))
+                        .Handler(new ActionChannelInitializer<IChannel>(channel =>
+                        {
+                            IChannelPipeline pipeline = channel.Pipeline;
 
-                    var cts = new CancellationTokenSource();
+                            pipeline.AddLast("handler", this);
-                    cts.CancelAfter(TimeoutTime);
+                        }));
-                    await Task.Run(() => _socketClient.Connect(_host, _port), cts.Token);
 
-                    if (_socketClient.Client.Connected)
+                    Channel = await bootstrap.BindAsync(IPEndPoint.MinPort);
+
+                    if (Channel.Active)
                     {
-                        _taskCancel = false;
                         Controller.SendStart();
-                        ReceiveMsgThreadStart();
                         logger.LogInformation("Udp client {ConnectionToken} connected", ConnectionToken);
                         return true;
                     }
@@ -138,7 +142,7 @@ namespace Modbus.Net
         /// <inheritdoc />
         public override bool Disconnect()
         {
-            if (_socketClient == null)
+            if (Channel == null)
                 return true;
 
             try
@@ -150,11 +154,14 @@ namespace Modbus.Net
             catch (Exception err)
             {
                 logger.LogError(err, "Udp client {ConnectionToken} disconnected exception", ConnectionToken);
+
+                RefreshErrorCount();
+
                 return false;
             }
             finally
             {
-                _socketClient = null;
+                Channel = null;
             }
         }
 
@@ -171,71 +178,53 @@ namespace Modbus.Net
                 RefreshSendCount();
 
                 logger.LogDebug("Udp client {ConnectionToken} send text len = {Length}", ConnectionToken, datagram.Length);
-                logger.LogDebug($"Udp client {ConnectionToken} send: {String.Concat(datagram.Select(p => " " + p.ToString("X2")))}");
+                logger.LogDebug($"Udp client {ConnectionToken} send: {string.Concat(datagram.Select(p => " " + p.ToString("X2")))}");
-                await _socketClient.SendAsync(datagram, datagram.Length);
+                IByteBuffer buffer = Unpooled.Buffer();
+                buffer.WriteBytes(datagram);
+                var packet = new DatagramPacket((IByteBuffer)buffer.Retain(), new IPEndPoint(IPAddress.Parse(_host), _port));
+                await Channel.WriteAndFlushAsync(packet);
             }
             catch (Exception err)
             {
                 logger.LogError(err, "Udp client {ConnectionToken} send exception", ConnectionToken);
+
+                RefreshErrorCount();
+
                 Dispose();
             }
         }
 
-        /// <inheritdoc />
+        /// <inheridoc />
-        protected override void ReceiveMsgThreadStart()
+        public override async void ChannelRead(IChannelHandlerContext ctx, object message)
-        {
-            _receiveThread = Task.Run(ReceiveMessage);
-        }
-
-        /// <inheritdoc />
-        protected override void ReceiveMsgThreadStop()
-        {
-            _taskCancel = true;
-        }
-
-        /// <summary>
-        ///     接收返回消息
-        /// </summary>
-        /// <returns>返回的消息</returns>
-        protected async Task ReceiveMessage()
         {
             try
             {
-                while (!_taskCancel)
+                if (message is DatagramPacket packet)
                 {
-                    if (_socketClient == null) break;
+                    var buffer = packet.Content;
-                    var receive = await _socketClient.ReceiveAsync();
+                    byte[] msg = buffer.Array.Slice(buffer.ArrayOffset, buffer.ReadableBytes);
-
+                    logger.LogDebug("Udp client {ConnectionToken} receive text len = {Length}", ConnectionToken,
-                    var len = receive.Buffer.Length;
+                                    msg.Length);
-                    // 异步接收回答
+                    logger.LogDebug(
-                    if (len > 0)
+                        $"Udp client {ConnectionToken} receive: {string.Concat(msg.Select(p => " " + p.ToString("X2")))}");
+                    var isMessageConfirmed = Controller.ConfirmMessage(msg);
+                    if (isMessageConfirmed != null)
                     {
-                        if (receive.Buffer.Clone() is byte[] receiveBytes)
+                        foreach (var confirmed in isMessageConfirmed)
                         {
-                            logger.LogDebug("Udp client {ConnectionToken} receive text len = {Length}", ConnectionToken,
+                            if (confirmed.Item2 == false)
-                                receiveBytes.Length);
-                            logger.LogDebug(
-                                $"Udp client {ConnectionToken} receive: {String.Concat(receiveBytes.Select(p => " " + p.ToString("X2")))}");
-                            var isMessageConfirmed = Controller.ConfirmMessage(receiveBytes);
-                            if (isMessageConfirmed != null)
                             {
-                                foreach (var confirmed in isMessageConfirmed)
+                                var sendMessage = InvokeReturnMessage(confirmed.Item1);
+                                //主动传输事件
+                                if (sendMessage != null)
                                 {
-                                    if (confirmed.Item2 == false)
+                                    await SendMsgWithoutConfirm(sendMessage);
-                                    {
-                                        var sendMessage = InvokeReturnMessage(confirmed.Item1);
-                                        //主动传输事件
-                                        if (sendMessage != null)
-                                        {
-                                            await SendMsgWithoutConfirm(sendMessage);
-                                        }
-                                    }
                                 }
                             }
                         }
-
-                        RefreshReceiveCount();
                     }
+
+                    RefreshReceiveCount();
                 }
             }
             catch (ObjectDisposedException)
@@ -245,7 +234,10 @@ namespace Modbus.Net
             catch (Exception err)
             {
                 logger.LogError(err, "Udp client {ConnectionToken} receive exception", ConnectionToken);
-                //CloseClientSocket();
+
+                RefreshErrorCount();
+
+                await CloseClientSocket();
             }
         }
 
@@ -267,25 +259,25 @@ namespace Modbus.Net
             logger.LogDebug("Udp client {ConnectionToken} error count: {ErrorCount}", ConnectionToken, _errorCount);
         }
 
-        private void CloseClientSocket()
+        private async Task CloseClientSocket()
         {
             try
             {
                 Controller.SendStop();
                 Controller.Clear();
-                ReceiveMsgThreadStop();
+                if (Channel != null)
-                if (_socketClient != null)
                 {
-                    if (_socketClient.Client?.Connected == true)
+                    if (Channel.Active)
                     {
-                        _socketClient.Client.Disconnect(false);
+                        await Channel.CloseAsync();
                     }
-                    _socketClient.Close();
                 }
             }
             catch (Exception ex)
             {
                 logger.LogError(ex, "Udp client {ConnectionToken} client close exception", ConnectionToken);
+
+                RefreshErrorCount();
             }
         }
     }

Modbus.Net/Modbus.Net/Modbus.Net.csproj

@@ -5,7 +5,7 @@
     <AssemblyName>Modbus.Net</AssemblyName>
     <RootNamespace>Modbus.Net</RootNamespace>
     <PackageId>Modbus.Net</PackageId>
-    <Version>1.4.0-beta04</Version>
+    <Version>1.4.1-beta05</Version>
     <Product>Modbus.Net</Product>
     <Authors>Chris L.(Luo Sheng)</Authors>
     <Company>Hangzhou Delian Science Technology Co.,Ltd.</Company>
@@ -29,6 +29,7 @@
   </PropertyGroup>
 
   <ItemGroup>
+    <PackageReference Include="DotNetty.Handlers" Version="0.7.5" />
     <PackageReference Include="Microsoft.Extensions.Configuration.Json" Version="7.0.0" />
     <PackageReference Include="Nito.AsyncEx" Version="5.1.2" />
     <PackageReference Include="Quartz" Version="3.6.2" />