dotnet/Sers.sln

@@ -157,6 +157,8 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Apm", "Apm", "{AF17FAEF-DBC
 EndProject
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Sers.Gover.Apm.Zipkin", "Sers\Sers.ServiceCenter\Apm\Sers.Gover.Apm.Zipkin\Sers.Gover.Apm.Zipkin.csproj", "{DD0CB14B-D8E6-46EC-AE42-2B1215C1BBDD}"
 EndProject
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Sers.CL.Ipc.NamedPipe", "Sers\Sers.CL\Ipc\Sers.CL.Ipc.NamedPipe\Sers.CL.Ipc.NamedPipe.csproj", "{545F1A7C-68B6-4F1F-B165-1A0F8F3DC51D}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -331,6 +333,10 @@ Global
 		{DD0CB14B-D8E6-46EC-AE42-2B1215C1BBDD}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{DD0CB14B-D8E6-46EC-AE42-2B1215C1BBDD}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{DD0CB14B-D8E6-46EC-AE42-2B1215C1BBDD}.Release|Any CPU.Build.0 = Release|Any CPU
+		{545F1A7C-68B6-4F1F-B165-1A0F8F3DC51D}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{545F1A7C-68B6-4F1F-B165-1A0F8F3DC51D}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{545F1A7C-68B6-4F1F-B165-1A0F8F3DC51D}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{545F1A7C-68B6-4F1F-B165-1A0F8F3DC51D}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
@@ -409,6 +415,7 @@ Global
 		{141C89FB-54A7-4283-8649-D6EF51B22FA9} = {DA14C04D-FEEF-4A3C-A9F2-BBA25897B9A4}
 		{AF17FAEF-DBCC-4E5C-954D-BD4F60A016C1} = {97C34C06-AE79-4E1A-8D04-EB0107643302}
 		{DD0CB14B-D8E6-46EC-AE42-2B1215C1BBDD} = {AF17FAEF-DBCC-4E5C-954D-BD4F60A016C1}
+		{545F1A7C-68B6-4F1F-B165-1A0F8F3DC51D} = {ECEEABBB-DEA2-4CD4-99EA-3E048DB23219}
 	EndGlobalSection
 	GlobalSection(ExtensibilityGlobals) = postSolution
 		SolutionGuid = {C7DA16E3-9949-49FA-B0B4-F830636DE60F}

dotnet/Sers/Sers.CL/Ipc/Sers.CL.Ipc.NamedPipe/DeliveryClient.cs

@@ -0,0 +1,86 @@
+using System;
+using System.IO.Pipes;
+using Sers.Core.CL.MessageDelivery;
+using Vit.Core.Module.Log;
+
+namespace Sers.CL.Ipc.NamedPipe
+{
+    public class DeliveryClient: IDeliveryClient
+    {
+
+        /// <summary>
+        /// 尝试连接超时时间（单位：毫秒）（默认：1000）
+        /// </summary>
+        public int connectTimeoutMs = 1000;
+
+        DeliveryConnection _conn  = new DeliveryConnection();
+        public IDeliveryConnection conn => _conn;
+
+        /// <summary>
+        /// 请勿处理耗时操作，需立即返回。接收到客户端的数据事件
+        /// </summary>
+        public Action<IDeliveryConnection, ArraySegment<byte>> Conn_OnGetFrame { set { _conn.OnGetFrame = value; } }
+
+        public Action<IDeliveryConnection> Conn_OnDisconnected { set=> _conn.Conn_OnDisconnected=value; }
+
+
+        /// <summary>
+        /// 默认 "."
+        /// </summary>
+        public string serverName = ".";
+        /// <summary>
+        /// 如： "Sers.CL.Ipc"
+        /// </summary>
+        public string pipeName = "Sers.CL.Ipc";
+ 
+
+        public bool Connect()
+        {
+            Logger.Info("[CL.DeliveryClient] Ipc.NamedPipe, connecting... serverName:" + serverName + " pipeName:" + pipeName);
+
+            NamedPipeClientStream client;
+            try
+            {
+
+                client = new NamedPipeClientStream(serverName, pipeName, PipeDirection.InOut, PipeOptions.Asynchronous);
+
+                client.Connect(connectTimeoutMs);
+
+                if (!client.IsConnected)
+                {
+                    return false;
+                }
+
+            }
+            catch (Exception ex)
+            {
+                //服务启动失败
+                Logger.Error("[CL.DeliveryClient] Ipc.NamedPipe, connect - Error", ex);
+                return false;
+            }
+             
+
+            _conn.Init(client);
+
+            _conn.StartBackThreadToReceiveMsg();
+            Logger.Info("[CL.DeliveryClient] Ipc.NamedPipe, connected.");
+            return true;
+        }
+
+        /// Disconnect from the host.
+        public void Close()
+        {
+            try
+            {
+                _conn?.Close();
+            }
+            catch (Exception ex)
+            {
+                Logger.Error(ex);
+            }
+            _conn = null;
+        }
+
+
+    }
+}

dotnet/Sers/Sers.CL/Ipc/Sers.CL.Ipc.NamedPipe/DeliveryConnection.cs

@@ -0,0 +1,222 @@
+using System;
+using System.Collections.Generic;
+using System.IO.Pipes;
+using System.Threading;
+using System.Threading.Tasks;
+using Sers.Core.CL.MessageDelivery;
+using Vit.Core.Module.Log;
+using Vit.Core.Util.Pipelines;
+using Vit.Core.Util.Pool;
+using Vit.Extensions;
+
+namespace Sers.CL.Ipc.NamedPipe
+{
+    /// <summary>
+    ///  
+    /// </summary>
+    public class DeliveryConnection : IDeliveryConnection
+    {
+
+        ~DeliveryConnection()
+        {
+            Close();
+        }
+
+        /// <summary>
+        /// 连接状态(0:waitForCertify; 2:certified; 4:waitForClose; 8:closed;)
+        /// </summary>
+        public byte state { get; set; } = DeliveryConnState.waitForCertify;
+
+
+        public Action<IDeliveryConnection> Conn_OnDisconnected { get; set; }
+
+        /// <summary>
+        /// 请勿处理耗时操作，需立即返回。接收到客户端的数据事件
+        /// </summary>
+        public Action<IDeliveryConnection, ArraySegment<byte>> OnGetFrame { private get; set; }
+
+
+        public void SendFrameAsync(List<ArraySegment<byte>> data)
+        {
+            if (data == null || stream == null) return;
+
+            if (!stream.IsConnected) 
+            {
+                Task.Run(Close);
+                return;
+            }
+            try
+            {
+                Int32 len = data.ByteDataCount();
+                data.Insert(0, len.Int32ToArraySegmentByte());
+
+                var bytes = data.ByteDataToBytes();
+
+                stream.WriteAsync(bytes, 0, bytes.Length);
+                stream.FlushAsync();
+            }
+            catch (Exception ex) when (!(ex.GetBaseException() is ThreadInterruptedException))
+            {
+                Logger.Error(ex);
+                Task.Run(Close);
+            }
+        }
+
+
+
+        public void Close()
+        {
+            if (state == DeliveryConnState.closed) return;
+            state = DeliveryConnState.closed;
+
+            try
+            {
+                stream.Close();
+                stream.Dispose();
+
+                //stream.Shutdown(SocketShutdown.Both);
+            }
+            catch (Exception ex)
+            {
+                Logger.Error(ex);
+            }
+
+            try
+            {
+                Conn_OnDisconnected?.Invoke(this);
+            }
+            catch (Exception ex)
+            {
+                Logger.Error(ex);
+            }
+        }
+
+
+        public void Init(PipeStream stream)
+        {
+            this.stream = stream;
+
+            connectTime = DateTime.Now;
+        }
+
+
+
+        #region taskToReceiveMsg       
+
+        class StreamReader
+        {
+            public PipeStream stream;
+
+
+            public Action<ArraySegment<byte>> OnReadData;
+            public Action OnClose;
+
+            //定义异步读取状态类
+            class AsyncState
+            {
+                public PipeStream stream { get; set; }
+                public byte[] buffer { get; set; }        
+            }
+
+            /// <summary>
+            /// 缓存区大小
+            /// </summary>
+            public int receiveBufferSize = 8 * 1024;
+
+            public void Start()
+            {
+                try
+                {
+                    var buffer = DataPool.BytesGet(receiveBufferSize);
+
+                    var asyncState = new AsyncState { stream = stream, buffer = buffer };
+
+                    //异步读取
+                    if (stream.IsConnected)
+                    {
+                        var result=stream.BeginRead(buffer, 0, receiveBufferSize, new AsyncCallback(AsyncReadCallback), asyncState);                  
+                        return;
+                    }
+                }
+                catch (Exception ex)
+                {
+                    Logger.Error(ex);
+                }
+                Task.Run(OnClose);
+            }
+
+
+            //异步读取回调处理方法
+            void AsyncReadCallback(IAsyncResult asyncResult)
+            {
+                try
+                {
+                    var asyncState = (AsyncState)asyncResult.AsyncState;
+                    int readCount = asyncState.stream.EndRead(asyncResult);
+
+                    if (readCount > 0)
+                    {
+                        //输出读取内容值
+                        OnReadData(new ArraySegment<byte>(asyncState.buffer, 0, readCount));
+
+                        asyncState.buffer = DataPool.BytesGet(receiveBufferSize);
+
+
+                        //再次执行异步读取操作
+                        if (asyncState.stream.IsConnected)
+                        {
+                            asyncState.stream.BeginRead(asyncState.buffer, 0, receiveBufferSize, new AsyncCallback(AsyncReadCallback), asyncState);
+                            return;
+                        }
+                    }
+                }
+                catch (Exception ex)
+                {
+                    Logger.Error(ex);
+                }
+                Task.Run(OnClose);
+            }
+        }
+
+
+        PipeFrame pipe = new PipeFrame() { OnDequeueData = ArraySegmentByteExtensions.ReturnToPool };
+
+        public void AppendData(ArraySegment<byte> data)
+        {
+            pipe.Write(data);
+
+            while (pipe.TryRead_SersFile(out var msgFrame))
+            {
+                OnGetFrame.Invoke(this, msgFrame);
+            }
+        }
+        public void StartBackThreadToReceiveMsg()
+        {
+            new StreamReader
+            {
+                stream = stream,
+                OnReadData = AppendData,
+                OnClose=Close
+            }.Start();
+        }
+
+ 
+        #endregion
+
+
+
+        /// <summary>
+        /// 通信对象
+        /// </summary>
+        public PipeStream stream { get; private set; }
+
+
+
+        /// <summary>
+        /// 连接时间
+        /// </summary>
+        public DateTime connectTime { get; private set; }
+
+
+    }
+}

dotnet/Sers/Sers.CL/Ipc/Sers.CL.Ipc.NamedPipe/DeliveryServer.cs

@@ -0,0 +1,198 @@
+using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.IO;
+using System.IO.Pipes;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using Sers.Core.CL.MessageDelivery;
+using Vit.Core.Module.Log;
+using Vit.Core.Util.Threading;
+
+namespace Sers.CL.Ipc.NamedPipe
+{
+    public class DeliveryServer: IDeliveryServer
+    {
+
+        /// <summary>
+        /// 如： "Sers.CL.Ipc"
+        /// </summary>
+        public string pipeName = "Sers.CL.Ipc";
+
+
+        public Action<IDeliveryConnection> Conn_OnDisconnected { private get; set; }
+        public Action<IDeliveryConnection> Conn_OnConnected { private get; set; } 
+
+
+        /// <summary>
+        ///  connHashCode -> DeliveryConnection
+        /// </summary>
+        readonly ConcurrentDictionary<int, DeliveryConnection> connMap = new ConcurrentDictionary<int, DeliveryConnection>();
+
+        public IEnumerable<IDeliveryConnection> ConnectedList => connMap.Values.Select(conn => ((IDeliveryConnection)conn));
+
+
+        LongTaskHelp tcpListenerAccept_BackThread = new LongTaskHelp();
+
+
+        #region Start     
+
+
+        /// <summary>
+        /// 启动服务
+        /// </summary>
+        public bool Start()
+        {
+            try
+            {
+                Logger.Info("[CL.DeliveryServer] Ipc.NamedPipe, starting... pipeName:" + pipeName);
+
+                #region (x.1)检测命名管道是否已经在使用
+                try
+                {
+
+                    if (File.Exists("\\\\.\\pipe\\" + pipeName)) 
+                    {
+                        Logger.Info("[CL.DeliveryServer] Ipc.NamedPipe, not started.pipeName already exists!");
+                        return false;
+                    }
+
+                    //String[] listOfPipes = System.IO.Directory.GetFiles(@"\\.\pipe\");
+                    //foreach(var t in listOfPipes) 
+                    //{
+                    //    Logger.Info("pipeName:   " + t);
+                    //}
+
+                    //using (var client = new NamedPipeClientStream(".", pipeName, PipeDirection.InOut, PipeOptions.Asynchronous))
+                    //{
+                    //    client.Connect(100);
+                    //    if (client.IsConnected)
+                    //    {
+                    //        Logger.Info("[CL.DeliveryServer] Ipc.NamedPipe, not started.pipeName already exists!");
+                    //        return false;
+                    //    }
+                    //}
+                }
+                catch (Exception ex)
+                {
+                    Logger.Error(ex);
+                }               
+                #endregion
+
+
+
+
+                #region (x.2)启动Task监听listener
+                tcpListenerAccept_BackThread.action = () =>
+                {
+                    try
+                    {
+                        while (true)
+                        {
+                            NamedPipeServerStream server = new NamedPipeServerStream(pipeName, PipeDirection.InOut, NamedPipeServerStream.MaxAllowedServerInstances,PipeTransmissionMode.Byte,PipeOptions.Asynchronous);
+
+                            //Console.WriteLine($"[{id}]服务器管道已创建。");
+
+                            // 等待客户端的连接
+                            server.WaitForConnection();
+
+                            Delivery_OnConnected(server);
+                        }
+                    }
+                    catch (Exception ex) when (!(ex.GetBaseException() is ThreadInterruptedException))
+                    {
+                        Logger.Error(ex);
+                    }
+                    finally
+                    {
+                        Stop();
+                    }
+                };
+                tcpListenerAccept_BackThread.Start();
+                #endregion               
+
+                Logger.Info("[CL.DeliveryServer] Ipc.NamedPipe, started.");
+                return true;
+            }
+            catch (Exception ex)
+            {
+                Logger.Error(ex);
+            }
+            return false;         
+
+        }
+
+        #endregion
+
+
+        #region Stop
+
+      
+
+        /// <summary>
+        /// 停止服务
+        /// </summary>
+        public void Stop()
+        {
+            //(x.1) stop conn
+            ConnectedList.ToList().ForEach(Delivery_OnDisconnected);            
+            connMap.Clear();
+
+            //(x.2) close
+            Task.Run(() =>
+            {
+                Logger.Info("[CL.DeliveryServer] Ipc.NamedPipe, stop...");       
+
+                tcpListenerAccept_BackThread.Stop();                 
+
+                Logger.Info("[CL.DeliveryServer] Ipc.NamedPipe, stoped");
+
+            });
+
+        }
+        #endregion
+
+
+        #region Delivery_Event
+
+
+        private DeliveryConnection Delivery_OnConnected(PipeStream client)
+        {
+            var conn = new DeliveryConnection();
+            conn.Init(client);
+        
+            conn.Conn_OnDisconnected = Delivery_OnDisconnected; 
+            connMap[conn.GetHashCode()] = conn;
+            try
+            {
+                Conn_OnConnected?.Invoke(conn);
+            }
+            catch (Exception ex)
+            {
+                Logger.Error(ex);
+            }
+
+            conn.StartBackThreadToReceiveMsg();
+
+            return conn;
+        }
+
+        private void Delivery_OnDisconnected(IDeliveryConnection _conn)
+        { 
+            var conn = (DeliveryConnection)_conn; 
+
+            connMap.TryRemove(conn.GetHashCode(), out _);
+
+            try
+            {
+                Conn_OnDisconnected?.Invoke(conn);
+            }
+            catch (Exception ex)
+            {
+                Logger.Error(ex);
+            }
+        }
+        #endregion
+    }
+}

dotnet/Sers/Sers.CL/Ipc/Sers.CL.Ipc.NamedPipe/Doc/DeliveryConnection_ThreadWait.cs

@@ -0,0 +1,238 @@
+using System;
+using System.Collections.Generic;
+using System.IO;
+using System.Net.Sockets;
+using System.Threading;
+using System.Threading.Tasks;
+using Sers.Core.CL.MessageDelivery;
+using Vit.Core.Module.Log;
+using Vit.Core.Util.Pool;
+using Vit.Core.Util.Threading;
+using Vit.Extensions;
+using static Sers.CL.Ipc.NamedPipe.DeliveryConnection1;
+
+namespace Sers.CL.Ipc.NamedPipe
+{
+    /// <summary>
+    ///  
+    /// </summary>
+    public class DeliveryConnection1 : IDeliveryConnection
+    {
+
+        ~DeliveryConnection1()
+        {
+            Close();
+        }
+
+        /// <summary>
+        /// 连接状态(0:waitForCertify; 2:certified; 4:waitForClose; 8:closed;)
+        /// </summary>
+        public byte state { get; set; } = DeliveryConnState.waitForCertify;
+
+
+        public Action<IDeliveryConnection> Conn_OnDisconnected { get; set; }
+
+        /// <summary>
+        /// 请勿处理耗时操作，需立即返回。接收到客户端的数据事件
+        /// </summary>
+        public Action<IDeliveryConnection, ArraySegment<byte>> OnGetFrame { private get; set; }
+
+
+        public void SendFrameAsync(List<ArraySegment<byte>> data)
+        {
+            if (data == null || socket == null) return;
+            try
+            {
+                Int32 len = data.ByteDataCount();
+                data.Insert(0, len.Int32ToArraySegmentByte());
+
+                var bytes = data.ByteDataToBytes();
+
+                socket.WriteAsync(bytes, 0, bytes.Length);
+ 
+            }
+            catch (Exception ex)
+            {
+                Logger.Error(ex);
+                Close();
+            }
+        }
+
+
+
+        public void Close()
+        {
+            if (socket == null) return;
+
+            state = DeliveryConnState.closed;
+
+            var socket_ = socket;
+            socket = null;
+            try
+            {
+                taskToReceiveMsg.Stop();
+            }
+            catch (Exception ex)
+            {
+                Logger.Error(ex);
+            }
+
+            try
+            {
+                socket_.Close();
+                socket_.Dispose();
+
+                //socket_.Shutdown(SocketShutdown.Both);
+            }
+            catch (Exception ex)
+            {
+                Logger.Error(ex);
+            }
+
+
+
+            try
+            {
+                Conn_OnDisconnected?.Invoke(this);
+            }
+            catch (Exception ex)
+            {
+                Logger.Error(ex);
+            }
+        }
+        public void Init(Stream stream)
+        {
+            this.socket = stream;
+
+            connectTime = DateTime.Now;
+        }
+
+
+
+        #region taskToReceiveMsg       
+
+        LongTaskHelp taskToReceiveMsg = new LongTaskHelp();
+        public void StartBackThreadToReceiveMsg()
+        {
+            taskToReceiveMsg.Stop();
+
+            taskToReceiveMsg.threadName = "Sers.CL.Socket.ThreadWait-taskToReceiveMsg";
+            taskToReceiveMsg.threadCount = 1;
+            taskToReceiveMsg.action = TaskToReceiveMsg;
+            taskToReceiveMsg.Start();
+        }
+
+        void TaskToReceiveMsg()
+        {
+            while (socket != null)
+            {
+                try
+                {
+                    while (socket != null)
+                    {
+                        OnGetFrame(this, ReadMsg());
+                    }
+                }
+                catch (Exception ex) when (!(ex.GetBaseException() is ThreadInterruptedException))
+                {
+                    Logger.Error(ex);
+                }
+            }
+        }
+        #endregion
+
+
+
+        /// <summary>
+        /// 通信SOCKET
+        /// </summary>
+        public Stream socket { get; private set; }
+
+
+
+        /// <summary>
+        /// 连接时间
+        /// </summary>
+        private DateTime connectTime { get; set; }
+
+
+
+
+
+
+        #region (x.x) socket层 封装 ReadMsg 
+        //线程不安全
+
+        /*
+            消息块格式：
+	            第一部分(len)    数据长度，4字节 Int32类型
+	            第二部分(data)   原始数据，长度由第二部分指定 
+             
+        */
+
+
+
+
+        internal ArraySegment<byte> ReadMsg()
+        {
+            #region Method Receive
+            void Receive(ArraySegment<byte> data)
+            {
+                int readedCount = 0;
+                int curCount;
+                do
+                {
+                    curCount = socket.Read(data.Array, data.Offset + readedCount, data.Count - readedCount);
+                    if (curCount == 0)
+                    {
+                        Logger.Error("[lith_190807_002]socket is closed.");
+                        throw new Exception("[lith_190418_002]socket is closed.");
+                    }
+                    readedCount += curCount;
+
+                } while (readedCount < data.Count);
+            }
+
+            #endregion
+
+
+            try
+            {
+
+                var bLen = DataPool.ArraySegmentByteGet(4);
+
+                //(x.1)获取 第一部分(len)  
+                Receive(bLen);
+                Int32 len = bLen.ArraySegmentByteToInt32();
+
+                //(x.2)获取第二部分(data)  
+                if (len < 0)
+                {
+                    Logger.Error("[lith_190807_003]socket read error.");
+                    throw new Exception("[lith_190505_001]socket read error.");
+                }
+                if (len == 0)
+                {
+                    return ArraySegmentByteExtensions.Null;
+                }
+                var data = DataPool.ArraySegmentByteGet(len);
+                Receive(data);
+                return data;
+            }
+            catch (Exception ex) when (!(ex.GetBaseException() is ThreadInterruptedException))
+            {
+                //连接断开
+                //Task.Run((Action)Close);
+                Close();
+                throw;
+            }
+        }
+
+
+        #endregion
+
+
+
+
+    }
+}

dotnet/Sers/Sers.CL/Ipc/Sers.CL.Ipc.NamedPipe/Doc/Pipe1.cs

@@ -0,0 +1,191 @@
+using System;
+using System.Collections.Generic;
+using System.IO;
+using System.IO.Pipes;
+using System.Security.Principal;
+using System.Threading;
+using System.Threading.Tasks;
+using Vit.Core.Module.Log;
+
+namespace DeliveryTest
+{
+    /// <summary>
+    /// https://www.liujiajia.me/2020/3/27/dotnet-core-named-pipe
+    /// </summary>
+    class Pipe1
+    {
+
+   
+        static void Main1(string[] args)
+        {
+            Logger.OnLog = (level, msg) => { Console.Write("[" + level + "]" + msg); };
+
+
+            Task.Run(StartServer);
+            //Task.Run(StartServer);
+            //Task.Run(StartServer);
+            //Task.Run(StartServer);
+
+
+            Task.Run(StartClient);
+            //Task.Run(StartClient);
+            //Task.Run(StartClient);
+            //Task.Run(StartClient);
+
+
+
+            while (true)
+            {
+                Thread.Sleep(5000);
+            }
+
+        }
+        static string pipeName = "demo2345";
+
+
+       static int serverId = 1;
+        static void StartServer()
+        {
+            string id = "server" + (serverId++);
+
+            while (1 == 1)
+            {
+
+                using (NamedPipeServerStream server = new NamedPipeServerStream(pipeName, PipeDirection.InOut))
+                {
+                    Console.WriteLine($"[{id}]服务器管道已创建。");
+
+                    // 等待客户端的连接
+                    server.WaitForConnection();
+                    Console.WriteLine($"[{id}]正在等待客户端连接......");
+
+
+                    Task.Run(()=> {
+                        while (true)
+                        {
+                         
+                            var buff1 = new byte[4];
+                            var len = server.Read(buff1, 0, 4);
+
+                        }
+
+                    });
+
+                    var buff = new byte[4] { 1,2,3,4};
+                    while (true)
+                    {
+                        //var len = server.Read(buff, 0, 4);
+                        //server.Write(buff,0,4);
+                        Thread.Sleep(1000);
+                    }
+
+                    using (StreamReader reader = new StreamReader(server))
+                    using (StreamWriter writer = new StreamWriter(server))
+                    {
+                    
+                        writer.AutoFlush = true;
+
+                        while (true) 
+                        {
+                            var msg = reader.ReadLine();
+                            writer.WriteLine("[]" + msg);
+                            //writer.Flush();
+                            Thread.Sleep(1000);
+                        }
+                      
+
+                        //
+
+                        //while (true)
+                        //{
+
+
+                        //string msg = "你好from-" + id;
+                        //writer.WriteLine(msg);
+                        //Thread.Sleep(1000);
+                        //}
+                    }
+
+                    server.Close();
+                }
+            }
+
+        }
+
+
+        static int clientId = 1;
+        static void StartClient()
+        {
+            string id = "client" + (clientId++);
+
+            try
+            {
+                using (NamedPipeClientStream client = new NamedPipeClientStream(".", pipeName, PipeDirection.InOut, PipeOptions.Asynchronous))
+                {
+                    Console.WriteLine($"[{id}]即将连接服务器。");
+                    client.Connect();
+                    Console.WriteLine($"[{id}]连接成功。");
+
+                    Task.Run(() => {
+                        while (true)
+                        {
+
+                            var buff1 = new byte[4];
+                            var len = client.Read(buff1, 0, 4);
+
+                        }
+
+                    });
+
+                    var buff = new byte[4] { 2, 2, 3, 4 };
+                    while (true)
+                    {
+                        //var len = server.Read(buff, 0, 4);
+                        client.Write(buff, 0, 4);
+                        client.Flush();
+                        Thread.Sleep(1000);
+                    }
+
+
+                    //var buff = new byte[4] { 4,3,2,1};
+                    //while (true)
+                    //{
+                    //    client.Write(buff, 0, 4);
+
+                    //    var buff2 = new byte[4] ;
+                    //    var len = client.Read(buff2, 0, 4);
+                    //    Thread.Sleep(1000);
+                    //}
+
+                    using (StreamWriter writer = new StreamWriter(client))
+                    using (StreamReader reader = new StreamReader(client))
+                    {
+                        //writer.AutoFlush = true;
+
+                        while (true)
+                        {
+                            string s = "hello world!";
+                            writer.WriteLine(s);
+
+                            var ss =   reader.ReadLine();
+                            Thread.Sleep(1000);
+
+                        }
+                        //string msg = null;
+                        //while ((msg = reader.ReadLine()) != null)
+                        //{
+                        //    Console.WriteLine($"[{id}]get message：{msg}");
+                        //}
+                    }
+
+                }
+            }
+            catch (Exception ex)
+            {
+                Console.WriteLine($"发生了异常：{ex}");
+            }
+
+        }
+
+    }
+}

dotnet/Sers/Sers.CL/Ipc/Sers.CL.Ipc.NamedPipe/Doc/Pipe2.cs

@@ -0,0 +1,193 @@
+using System;
+using System.Collections.Generic;
+using System.IO;
+using System.IO.Pipes;
+using System.Security.Principal;
+using System.Threading;
+using System.Threading.Tasks;
+using Vit.Core.Module.Log;
+
+namespace DeliveryTest
+{
+    /// <summary>
+    /// https://www.cnblogs.com/kasimlz/p/8619471.html
+    /// </summary>
+    class Pipe2
+    {
+
+   
+        static void Main2(string[] args)
+        {
+            Logger.OnLog = (level, msg) => { Console.Write("[" + level + "]" + msg); };
+
+
+            Task.Run(StartServer);
+            //Task.Run(StartServer);
+            //Task.Run(StartServer);
+            //Task.Run(StartServer);
+
+
+            Task.Run(StartClient);
+            //Task.Run(StartClient);
+            //Task.Run(StartClient);
+            //Task.Run(StartClient);
+
+
+
+            while (true)
+            {
+                Thread.Sleep(5000);
+            }
+
+        }
+        static string pipeName = "demo2345";
+
+
+       static int serverId = 1;
+        static void StartServer()
+        {
+            string id = "server" + (serverId++);
+
+            while (1 == 1)
+            {
+
+                using (NamedPipeServerStream server = new NamedPipeServerStream(pipeName, PipeDirection.InOut))
+                {
+                    Console.WriteLine($"[{id}]服务器管道已创建。");
+
+                    // 等待客户端的连接
+                    server.WaitForConnection();
+                    Console.WriteLine($"[{id}]正在等待客户端连接......");
+
+
+                  
+                    using (StreamReader reader = new StreamReader(server))
+                    using (StreamWriter writer = new StreamWriter(server))
+                    {
+                    
+                        writer.AutoFlush = true;
+
+                        Task.Run(() => {
+                            while (true)
+                            {
+                                var msg = reader.ReadLine();
+                                //var buff1 = new byte[4];
+                                //var len = server.Read(buff1, 0, 4);
+
+                            }
+
+                        });
+
+                        var buff = new byte[4] { 1, 2, 3, 4 };
+                        while (true)
+                        {
+                            //var len = server.Read(buff, 0, 4);
+                            //server.Write(buff,0,4);
+                            Thread.Sleep(1000);
+                        }
+
+
+
+                        while (true) 
+                        {
+                            var msg = reader.ReadLine();
+                            writer.WriteLine("[]" + msg);
+                            //writer.Flush();
+                            Thread.Sleep(1000);
+                        }
+                      
+
+                        //
+
+                        //while (true)
+                        //{
+
+
+                        //string msg = "你好from-" + id;
+                        //writer.WriteLine(msg);
+                        //Thread.Sleep(1000);
+                        //}
+                    }
+
+                    server.Close();
+                }
+            }
+
+        }
+
+
+        static int clientId = 1;
+        static void StartClient()
+        {
+            string id = "client" + (clientId++);
+
+            try
+            {
+                using (NamedPipeClientStream client = new NamedPipeClientStream(".", pipeName, PipeDirection.InOut, PipeOptions.None,TokenImpersonationLevel.None))
+                {
+                    Console.WriteLine($"[{id}]即将连接服务器。");
+                    client.Connect();
+                    Console.WriteLine($"[{id}]连接成功。");
+
+                  
+
+
+                    //var buff = new byte[4] { 4,3,2,1};
+                    //while (true)
+                    //{
+                    //    client.Write(buff, 0, 4);
+
+                    //    var buff2 = new byte[4] ;
+                    //    var len = client.Read(buff2, 0, 4);
+                    //    Thread.Sleep(1000);
+                    //}
+
+                    using (StreamWriter writer = new StreamWriter(client))
+                    using (StreamReader reader = new StreamReader(client))
+                    {
+                        writer.AutoFlush = true;
+                        //Task.Run(() => {
+                        //    while (true)
+                        //    {
+
+                        //        var ss = reader.ReadLine();
+
+                        //    }
+
+                        //});
+
+                        var buff = new byte[4] { 2, 2, 3, 4 };
+                        while (true)
+                        {
+                            //var len = server.Read(buff, 0, 4);
+                            string s = "hello world!";
+                            writer.WriteLine(s);
+                            Thread.Sleep(1000);
+                        }
+                        while (true)
+                        {
+                            string s = "hello world!";
+                            writer.WriteLine(s);
+
+                            var ss =   reader.ReadLine();
+                            Thread.Sleep(1000);
+
+                        }
+                        //string msg = null;
+                        //while ((msg = reader.ReadLine()) != null)
+                        //{
+                        //    Console.WriteLine($"[{id}]get message：{msg}");
+                        //}
+                    }
+
+                }
+            }
+            catch (Exception ex)
+            {
+                Console.WriteLine($"发生了异常：{ex}");
+            }
+
+        }
+
+    }
+}

dotnet/Sers/Sers.CL/Ipc/Sers.CL.Ipc.NamedPipe/Doc/Pipe3.cs

@@ -0,0 +1,268 @@
+using System;
+using System.Collections.Generic;
+using System.IO;
+using System.IO.Pipes;
+using System.Security.Principal;
+using System.Threading;
+using System.Threading.Tasks;
+using Vit.Core.Module.Log;
+
+namespace DeliveryTest
+{
+    /// <summary>
+    ///    BeginRead  https://blog.csdn.net/xutingzhou/article/details/8184266
+    /// </summary>
+    class Pipe3
+    {
+
+        class StreamReader 
+        {
+            public PipeStream stream;
+
+
+            public Action<byte[]> OnReadData;
+
+
+            //定义异步读取状态类
+            class AsyncState
+            {
+                public PipeStream FS { get; set; }
+                public byte[] Buffer { get; set; }
+                public ManualResetEvent EvtHandle { get; set; }
+            }
+            static int bufferSize = 512; 
+
+            public void Start() 
+            {
+                var buffer = new byte[bufferSize];
+                //构造BeginRead需要传递的状态
+                var asyncState = new AsyncState { FS = stream, Buffer = buffer, EvtHandle = new ManualResetEvent(false) };
+                //异步读取
+                var asyncResult = stream.BeginRead(buffer, 0, bufferSize, new AsyncCallback(AsyncReadCallback), asyncState);                
+            }
+ 
+
+            //异步读取回调处理方法
+            void AsyncReadCallback(IAsyncResult asyncResult)
+            {
+                var asyncState = (AsyncState)asyncResult.AsyncState;
+                int readCn = asyncState.FS.EndRead(asyncResult);
+                //判断是否读到内容
+                if (readCn > 0)
+                {
+                    byte[] buffer;
+                    if (readCn == bufferSize) buffer = asyncState.Buffer;
+                    else
+                    {
+                        buffer = new byte[readCn];
+                        Array.Copy(asyncState.Buffer, 0, buffer, 0, readCn);
+                    }
+                    //输出读取内容值
+                    OnReadData(buffer);
+                }
+
+                asyncState.Buffer = new byte[bufferSize];
+                //再次执行异步读取操作
+                asyncResult = asyncState.FS.BeginRead(asyncState.Buffer, 0, bufferSize, new AsyncCallback(AsyncReadCallback), asyncState);
+            }
+     
+
+
+        }
+
+
+        class StreamWriter
+        {
+            public PipeStream stream;
+
+
+            public Action<byte[]> OnReadData;
+
+
+            //定义异步读取状态类
+            class AsyncState
+            {
+                public PipeStream FS { get; set; }
+                public byte[] Buffer { get; set; }
+                public ManualResetEvent EvtHandle { get; set; }
+            }
+       
+
+            public void Write(byte[] buffer)
+            {
+             
+                //构造BeginRead需要传递的状态
+                var asyncState = new AsyncState { FS = stream, Buffer = buffer, EvtHandle = new ManualResetEvent(false) };
+                //异步读取
+                var asyncResult = stream.BeginWrite(buffer, 0, buffer.Length, new AsyncCallback(AsyncReadCallback), asyncState);
+            }
+
+
+            //异步读取回调处理方法
+            void AsyncReadCallback(IAsyncResult asyncResult)
+            {
+                var asyncState = (AsyncState)asyncResult.AsyncState;
+                asyncState.FS.EndWrite(asyncResult);
+            }
+
+
+
+        }
+
+
+
+        static void Main1(string[] args)
+        {
+            Logger.OnLog = (level, msg) => { Console.Write("[" + level + "]" + msg); };
+
+
+            Task.Run(StartServer);
+            //Task.Run(StartServer);
+            //Task.Run(StartServer);
+            //Task.Run(StartServer);
+
+
+            Task.Run(StartClient);
+            //Task.Run(StartClient);
+            //Task.Run(StartClient);
+            //Task.Run(StartClient);
+
+
+
+            while (true)
+            {
+                Thread.Sleep(5000);
+            }
+
+        }
+        static string pipeName = "demo2345";
+
+
+       static int serverId = 1;
+        static void StartServer()
+        {
+            string id = "server" + (serverId++);
+
+            while (1 == 1)
+            {
+
+                using (NamedPipeServerStream server = new NamedPipeServerStream(pipeName, PipeDirection.InOut))
+                {
+                    Console.WriteLine($"[{id}]服务器管道已创建。");
+
+                    // 等待客户端的连接
+                    server.WaitForConnection();
+                    Console.WriteLine($"[{id}]正在等待客户端连接......");
+
+
+                    new StreamReader { stream = server, OnReadData = (byte[] data) => {
+
+                        Console.WriteLine("getdata");
+
+                    } }.Start();
+
+
+                    var writer = new StreamWriter
+                    {
+                        stream = server
+
+                    };
+
+
+                    var buff = new byte[4] { 2, 2, 3, 4 };
+                    while (true)
+                    {
+                        writer.Write(buff);
+                        //var len = server.Read(buff, 0, 4);
+                        //client.Write(buff, 0, 4);
+                        //client.Flush();
+                        Thread.Sleep(1000);
+                    }
+
+
+
+                    //var buff = new byte[4] { 1,2,3,4};
+                    //while (true)
+                    //{
+                    //    //var len = server.Read(buff, 0, 4);
+                    //    //server.Write(buff,0,4);
+                    //    Thread.Sleep(1000);
+                    //}
+
+
+                    server.Close();
+                }
+            }
+
+        }
+
+
+        static int clientId = 1;
+        static void StartClient()
+        {
+            string id = "client" + (clientId++);
+
+            try
+            {
+                using (NamedPipeClientStream client = new NamedPipeClientStream(".", pipeName, PipeDirection.InOut, PipeOptions.Asynchronous))
+                {
+                    Console.WriteLine($"[{id}]即将连接服务器。");
+                    client.Connect();
+                    Console.WriteLine($"[{id}]连接成功。");
+
+                    //Task.Run(() => {
+                    //    while (true)
+                    //    {
+
+                    //        var buff1 = new byte[4];
+                    //        var len = client.Read(buff1, 0, 4);
+
+                    //    }
+
+                    //});
+
+
+                    new StreamReader
+                    {
+                        stream = client,
+                        OnReadData = (byte[] data) => {
+
+                            Console.WriteLine("getdata from server");
+
+                        }
+                    }.Start();
+
+
+
+
+
+                    var writer=new StreamWriter
+                    {
+                        stream = client
+                         
+                    } ;
+
+
+                    var buff = new byte[4] { 2, 2, 3, 4 };
+                    while (true)
+                    {
+                        writer.Write(buff);
+                        //var len = server.Read(buff, 0, 4);
+                        //client.Write(buff, 0, 4);
+                        //client.Flush();
+                        Thread.Sleep(1000);
+                    }
+
+                     
+
+                }
+            }
+            catch (Exception ex)
+            {
+                Console.WriteLine($"发生了异常：{ex}");
+            }
+
+        }
+
+    }
+}

dotnet/Sers/Sers.CL/Ipc/Sers.CL.Ipc.NamedPipe/Doc/Sers.CL.Ipc.NamedPipe - 额外命名管道/ConnectionKeyHelp.cs

@@ -0,0 +1,88 @@
+using System;
+using System.Collections.Generic;
+using System.IO;
+using System.IO.Pipes;
+using System.Text;
+using System.Threading;
+using Vit.Core.Module.Log;
+using Vit.Core.Util.Threading;
+
+namespace Sers.CL.Ipc.NamedPipe
+{
+    public static class ConnectionKeyHelp
+    {
+
+
+
+
+        public static void Publish(Func<string> beforeConnect, string pipeName = "Sers.Ipc")
+        {
+
+            using (NamedPipeServerStream server = new NamedPipeServerStream(pipeName, PipeDirection.Out))
+            {
+                //Console.WriteLine($"[{id}]服务器管道已创建。");
+
+                // 等待客户端的连接
+                server.WaitForConnection();
+
+                //Console.WriteLine($"[{id}]正在等待客户端连接......");
+
+                var connKey = beforeConnect();
+                Logger.Info("[Ipc]服务端已创建，connKey：" + connKey);
+
+                using (StreamWriter writer = new StreamWriter(server))
+                {
+                    writer.AutoFlush = true;
+                    writer.WriteLine(connKey);
+                }
+                server.Close();
+            }
+
+        }
+
+
+
+
+
+
+
+        /// <summary>
+        ///  
+        /// </summary>
+        /// <param name="pipeName"></param>
+        /// <param name="serverName"></param>
+        /// <param name="msTimeout"></param>
+        /// <returns> connectionKey </returns>
+        public static string Subscribe(string pipeName,string serverName=".",int msTimeout=10000) 
+            {
+                using (NamedPipeClientStream client = new NamedPipeClientStream(serverName, pipeName, PipeDirection.In))
+                {
+                    //Console.WriteLine($"[{id}]即将连接服务器。");
+
+                    var task=client.ConnectAsync();
+
+                    task.Wait(msTimeout);
+
+                    if (!task.IsCompleted) 
+                    {
+                        return null;
+                    }
+              
+
+                    using (StreamReader reader = new StreamReader(client))
+                    {
+                        string connectionKey = reader.ReadLine();
+                        return connectionKey;
+
+                    }
+                    //client.Close();
+                    return null;
+                }
+
+            }
+        
+       
+
+
+    }
+}

dotnet/Sers/Sers.CL/Ipc/Sers.CL.Ipc.NamedPipe/Doc/Sers.CL.Ipc.NamedPipe - 额外命名管道/DeliveryClient.cs

@@ -0,0 +1,96 @@
+using System;
+using System.IO.Pipes;
+using System.Net.Sockets;
+using Sers.Core.CL.MessageDelivery;
+using Vit.Core.Module.Log;
+
+namespace Sers.CL.Ipc.NamedPipe
+{
+    public class DeliveryClient: IDeliveryClient
+    {
+
+        DeliveryConnection _conn  = new DeliveryConnection();
+        public IDeliveryConnection conn => _conn;
+
+
+        /// <summary>
+        /// 请勿处理耗时操作，需立即返回。接收到客户端的数据事件
+        /// </summary>
+        public Action<IDeliveryConnection, ArraySegment<byte>> Conn_OnGetFrame { set { _conn.OnGetFrame = value; } }
+
+        public Action<IDeliveryConnection> Conn_OnDisconnected { set=> _conn.Conn_OnDisconnected=value; }
+
+
+
+
+
+
+        public string serverName = ".";
+        public string pipeName = "demo";
+
+
+        string connKey;
+
+        public bool Connect()
+        {
+            Logger.Info("[CL.DeliveryClient] Socket.ThreadWait,connecting... serverName:" + serverName + " pipeName:" + pipeName);
+     
+            try
+            {
+                connKey=ConnectionKeyHelp.Subscribe(pipeName, serverName);          
+                
+            }
+            catch (Exception ex)
+            {
+                //服务启动失败
+                Logger.Error("[CL.DeliveryClient] Socket.ThreadWait,connect - Error", ex);
+                return false;
+            }
+
+
+            NamedPipeClientStream client = new NamedPipeClientStream(serverName, pipeName + "." + connKey, PipeDirection.InOut, PipeOptions.Asynchronous);
+
+            //var task=client.ConnectAsync();
+
+            //task.Wait(10000);
+
+            //if (!task.IsCompleted)
+            //{
+            //    return false;
+            //}
+
+            client.Connect(10000);
+
+            if (!client.IsConnected) 
+            {
+                return false;
+            }
+
+            Logger.Info("[Ipc]客户端已创建，connKey：" + connKey);
+
+            _conn.Init(client);
+
+            _conn.StartBackThreadToReceiveMsg();
+            Logger.Info("[CL.DeliveryClient] Socket.ThreadWait,connected.");
+            return true;
+        }
+
+        /// Disconnect from the host.
+        public void Close()
+        {
+            try
+            {
+                _conn?.Close();
+            }
+            catch (Exception ex)
+            {
+                Logger.Error(ex);
+            }
+
+            _conn = null;
+
+        }
+
+
+    }
+}

dotnet/Sers/Sers.CL/Ipc/Sers.CL.Ipc.NamedPipe/Doc/Sers.CL.Ipc.NamedPipe - 额外命名管道/DeliveryConnection.cs

@@ -0,0 +1,241 @@
+using System;
+using System.Collections.Generic;
+using System.IO;
+using System.Net.Sockets;
+using System.Threading;
+using System.Threading.Tasks;
+using Sers.Core.CL.MessageDelivery;
+using Vit.Core.Module.Log;
+using Vit.Core.Util.Pool;
+using Vit.Core.Util.Threading;
+using Vit.Extensions;
+using static Sers.CL.Ipc.NamedPipe.DeliveryConnection;
+
+namespace Sers.CL.Ipc.NamedPipe
+{
+    /// <summary>
+    ///  
+    /// </summary>
+    public class DeliveryConnection : IDeliveryConnection
+    {
+
+        ~DeliveryConnection()
+        {
+            Close();
+        }
+
+        /// <summary>
+        /// 连接状态(0:waitForCertify; 2:certified; 4:waitForClose; 8:closed;)
+        /// </summary>
+        public byte state { get; set; } = DeliveryConnState.waitForCertify;
+
+
+        public Action<IDeliveryConnection> Conn_OnDisconnected { get; set; }
+
+        /// <summary>
+        /// 请勿处理耗时操作，需立即返回。接收到客户端的数据事件
+        /// </summary>
+        public Action<IDeliveryConnection, ArraySegment<byte>> OnGetFrame { private get; set; }
+
+
+        public void SendFrameAsync(List<ArraySegment<byte>> data)
+        {
+            if (data == null || socket == null) return;
+            try
+            {
+                Int32 len = data.ByteDataCount();
+                data.Insert(0, len.Int32ToArraySegmentByte());
+
+                var bytes = data.ByteDataToBytes();
+
+
+
+                socket.WriteAsync(bytes, 0, bytes.Length);
+
+                //socket.FlushAsync();
+            }
+            catch (Exception ex)
+            {
+                Logger.Error(ex);
+                Close();
+            }
+        }
+
+
+
+        public void Close()
+        {
+            if (socket == null) return;
+
+            state = DeliveryConnState.closed;
+
+            var socket_ = socket;
+            socket = null;
+            try
+            {
+                taskToReceiveMsg.Stop();
+            }
+            catch (Exception ex)
+            {
+                Logger.Error(ex);
+            }
+
+            try
+            {
+                socket_.Close();
+                socket_.Dispose();
+
+                //socket_.Shutdown(SocketShutdown.Both);
+            }
+            catch (Exception ex)
+            {
+                Logger.Error(ex);
+            }
+
+
+
+            try
+            {
+                Conn_OnDisconnected?.Invoke(this);
+            }
+            catch (Exception ex)
+            {
+                Logger.Error(ex);
+            }
+        }
+        public void Init(Stream stream)
+        {
+            this.socket = stream;
+
+            connectTime = DateTime.Now;
+        }
+
+
+
+        #region taskToReceiveMsg       
+
+        LongTaskHelp taskToReceiveMsg = new LongTaskHelp();
+        public void StartBackThreadToReceiveMsg()
+        {
+            taskToReceiveMsg.Stop();
+
+            taskToReceiveMsg.threadName = "Sers.CL.Socket.ThreadWait-taskToReceiveMsg";
+            taskToReceiveMsg.threadCount = 1;
+            taskToReceiveMsg.action = TaskToReceiveMsg;
+            taskToReceiveMsg.Start();
+        }
+
+        void TaskToReceiveMsg()
+        {
+            while (socket != null)
+            {
+                try
+                {
+                    while (socket != null)
+                    {
+                        OnGetFrame(this, ReadMsg());
+                    }
+                }
+                catch (Exception ex) when (!(ex.GetBaseException() is ThreadInterruptedException))
+                {
+                    Logger.Error(ex);
+                }
+            }
+        }
+        #endregion
+
+
+
+        /// <summary>
+        /// 通信SOCKET
+        /// </summary>
+        public Stream socket { get; private set; }
+
+
+
+        /// <summary>
+        /// 连接时间
+        /// </summary>
+        private DateTime connectTime { get; set; }
+
+
+
+
+
+
+        #region (x.x) socket层 封装 ReadMsg 
+        //线程不安全
+
+        /*
+            消息块格式：
+	            第一部分(len)    数据长度，4字节 Int32类型
+	            第二部分(data)   原始数据，长度由第二部分指定 
+             
+        */
+
+
+
+
+        internal ArraySegment<byte> ReadMsg()
+        {
+            #region Method Receive
+            void Receive(ArraySegment<byte> data)
+            {
+                int readedCount = 0;
+                int curCount;
+                do
+                {
+                    curCount = socket.Read(data.Array, data.Offset + readedCount, data.Count - readedCount);
+                    if (curCount == 0)
+                    {
+                        Logger.Error("[lith_190807_002]socket is closed.");
+                        throw new Exception("[lith_190418_002]socket is closed.");
+                    }
+                    readedCount += curCount;
+
+                } while (readedCount < data.Count);
+            }
+
+            #endregion
+
+
+            try
+            {
+
+                var bLen = DataPool.ArraySegmentByteGet(4);
+
+                //(x.1)获取 第一部分(len)  
+                Receive(bLen);
+                Int32 len = bLen.ArraySegmentByteToInt32();
+
+                //(x.2)获取第二部分(data)  
+                if (len < 0)
+                {
+                    Logger.Error("[lith_190807_003]socket read error.");
+                    throw new Exception("[lith_190505_001]socket read error.");
+                }
+                if (len == 0)
+                {
+                    return ArraySegmentByteExtensions.Null;
+                }
+                var data = DataPool.ArraySegmentByteGet(len);
+                Receive(data);
+                return data;
+            }
+            catch (Exception ex) when (!(ex.GetBaseException() is ThreadInterruptedException))
+            {
+                //连接断开
+                //Task.Run((Action)Close);
+                Close();
+                throw;
+            }
+        }
+
+
+        #endregion
+
+
+
+
+    }
+}

dotnet/Sers/Sers.CL/Ipc/Sers.CL.Ipc.NamedPipe/Doc/Sers.CL.Ipc.NamedPipe - 额外命名管道/DeliveryServer.cs

@@ -0,0 +1,189 @@
+using System;
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.IO;
+using System.IO.Pipes;
+using System.Linq;
+using System.Net;
+using System.Net.Sockets;
+using System.Threading;
+using System.Threading.Tasks;
+using Sers.Core.CL.MessageDelivery;
+using Vit.Core.Module.Log;
+using Vit.Core.Util.Common;
+using Vit.Core.Util.Net;
+using Vit.Core.Util.Threading;
+
+namespace Sers.CL.Ipc.NamedPipe
+{
+    public class DeliveryServer: IDeliveryServer
+    {
+
+     
+        public string pipeName = "demo";
+
+
+        public Action<IDeliveryConnection> Conn_OnDisconnected { private get; set; }
+        public Action<IDeliveryConnection> Conn_OnConnected { private get; set; }
+                           
+
+
+
+ 
+
+
+        /// <summary>
+        ///  connHashCode -> DeliveryConnection
+        /// </summary>
+        readonly ConcurrentDictionary<int, DeliveryConnection> connMap = new ConcurrentDictionary<int, DeliveryConnection>();
+
+        public IEnumerable<IDeliveryConnection> ConnectedList => connMap.Values.Select(conn => ((IDeliveryConnection)conn));
+
+
+        LongTaskHelp tcpListenerAccept_BackThread = new LongTaskHelp();
+
+
+        #region Start
+
+
+        string BeforeConnect() 
+        {
+            string connKey = CommonHelp.NewGuid();
+
+            Task.Run(() => {
+
+
+                var server = new NamedPipeServerStream(pipeName + "." + connKey, PipeDirection.InOut);
+                //TODO  10秒无连接 强制关闭
+                // 等待客户端的连接
+                server.WaitForConnection();
+
+                if(server.IsConnected)
+                Delivery_OnConnected(server);
+                //server.Close();
+
+            });
+            return connKey;
+        }
+
+
+        /// <summary>
+        /// 启动服务
+        /// </summary>
+        public bool Start()
+        {
+            try
+            {
+                
+
+                Logger.Info("[CL.DeliveryServer] Socket.ThreadWait,starting... pipeName:" + pipeName);
+ 
+
+
+                #region (x.2)启动Task监听listener
+                tcpListenerAccept_BackThread.action = () =>
+                {
+                    try
+                    {
+                        while (true)
+                        {
+                            ConnectionKeyHelp.Publish(BeforeConnect, pipeName);
+                        }
+                    }
+                    catch (Exception ex) when (!(ex.GetBaseException() is ThreadInterruptedException))
+                    {
+                        Logger.Error(ex);
+                    }
+                    finally
+                    {
+                        Stop();
+                    }
+                };
+                tcpListenerAccept_BackThread.Start();
+                #endregion               
+
+                Logger.Info("[CL.DeliveryServer] Socket.ThreadWait,started.");
+                return true;
+            }
+            catch (Exception ex)
+            {
+                Logger.Error(ex);
+            }
+            return false;         
+
+        }
+
+        #endregion
+
+
+        #region Stop
+
+        /// <summary>
+        /// 停止服务
+        /// </summary>
+        public void Stop()
+        {
+     
+
+            //(x.1) stop conn
+            ConnectedList.ToList().ForEach(Delivery_OnDisconnected);            
+            connMap.Clear();
+
+            //(x.2) close socket
+            Task.Run(() =>
+            {
+                Logger.Info("[CL.DeliveryServer] Socket.ThreadWait,stop...");       
+
+                tcpListenerAccept_BackThread.Stop();
+                 
+
+                Logger.Info("[CL.DeliveryServer] Socket.ThreadWait,stoped");
+
+            });
+
+        }
+        #endregion
+
+
+        #region Delivery_Event
+
+
+        private DeliveryConnection Delivery_OnConnected(Stream client)
+        {
+            var conn = new DeliveryConnection();
+            conn.Init(client);
+        
+            conn.Conn_OnDisconnected = Delivery_OnDisconnected; 
+            connMap[conn.GetHashCode()] = conn;
+            try
+            {
+                Conn_OnConnected?.Invoke(conn);
+            }
+            catch (Exception ex)
+            {
+                Logger.Error(ex);
+            }
+
+            conn.StartBackThreadToReceiveMsg();
+
+            return conn;
+        }
+
+        private void Delivery_OnDisconnected(IDeliveryConnection _conn)
+        { 
+            var conn = (DeliveryConnection)_conn; 
+
+            connMap.TryRemove(conn.GetHashCode(), out _);
+
+            try
+            {
+                Conn_OnDisconnected?.Invoke(conn);
+            }
+            catch (Exception ex)
+            {
+                Logger.Error(ex);
+            }
+        }
+        #endregion
+    }
+}

dotnet/Sers/Sers.CL/Ipc/Sers.CL.Ipc.NamedPipe/Doc/Sers.CL.Ipc.NamedPipe - 额外命名管道/Doc/Program.cs

@@ -0,0 +1,118 @@
+using System;
+using System.Collections.Generic;
+using System.IO;
+using System.IO.Pipes;
+using System.Threading;
+using System.Threading.Tasks;
+using Vit.Core.Module.Log;
+
+namespace DeliveryTest
+{
+    /// <summary>
+    /// https://www.liujiajia.me/2020/3/27/dotnet-core-named-pipe
+    /// </summary>
+    class Program
+    {
+
+   
+        static void Main(string[] args)
+        {
+            Logger.OnLog = (level, msg) => { Console.Write("[" + level + "]" + msg); };
+
+
+            Task.Run(StartServer);
+            //Task.Run(StartServer);
+            //Task.Run(StartServer);
+            //Task.Run(StartServer);
+
+
+            Task.Run(StartClient);
+            Thread.Sleep(5000);
+            Task.Run(StartClient);
+            //Task.Run(StartClient);
+            //Task.Run(StartClient);
+
+
+
+            while (true)
+            {
+                Thread.Sleep(5000);
+            }
+
+        }
+
+
+
+       static int serverId = 1;
+        static void StartServer()
+        {
+            string id = "server" + (serverId++);
+
+            while (1 == 1)
+            {
+
+                using (NamedPipeServerStream server = new NamedPipeServerStream("demo", PipeDirection.Out))
+                {
+                    Console.WriteLine($"[{id}]服务器管道已创建。");
+
+                    // 等待客户端的连接
+                    server.WaitForConnection();
+                    Console.WriteLine($"[{id}]正在等待客户端连接......");
+
+                    using (StreamWriter writer = new StreamWriter(server))
+                    {
+                        writer.AutoFlush = true;
+                        int t = 0;
+                        while (++t<10)
+                        {
+
+                            string msg = "你好from-" + id;
+                        writer.WriteLine(msg);
+                        Thread.Sleep(1000);
+                        }
+                    }
+
+                    //server.Close();
+                }
+            }
+
+        }
+
+
+        static int clientId = 1;
+        static void StartClient()
+        {
+            string id = "client" + (clientId++);
+
+            try
+            {
+                using (NamedPipeClientStream client = new NamedPipeClientStream(".", "demo", PipeDirection.In))
+                {
+                    Console.WriteLine($"[{id}]即将连接服务器。");
+
+
+                    client.Connect();
+                    Console.WriteLine($"[{id}]连接成功。");
+
+
+                    using (StreamReader reader = new StreamReader(client))
+                    {
+                        string msg = null;
+                        while ((msg = reader.ReadLine()) != null)
+                        {
+                            Console.WriteLine($"[{id}]get message：{msg}");
+                        }
+                    }
+
+                }
+            }
+            catch (Exception ex)
+            {
+                Console.WriteLine($"发生了异常：{ex}");
+            }
+            Console.WriteLine($"[{id}]closed");
+
+        }
+
+    }
+}

dotnet/Sers/Sers.CL/Ipc/Sers.CL.Ipc.NamedPipe/Doc/Sers.CL.Ipc.NamedPipe - 额外命名管道/OrganizeClientBuilder.cs

@@ -0,0 +1,21 @@
+using System.Collections.Generic;
+using Newtonsoft.Json.Linq;
+using Sers.Core.CL.MessageOrganize;
+using Sers.Core.CL.MessageOrganize.DefaultOrganize;
+using Vit.Extensions;
+
+namespace Sers.CL.Ipc.NamedPipe
+{
+    public class OrganizeClientBuilder : IOrganizeClientBuilder
+    {
+        public void Build(List<IOrganizeClient> organizeList, JObject config)
+        {
+            var delivery = new DeliveryClient();
+
+            delivery.serverName = config["serverName"].ConvertToString();
+            delivery.pipeName = config["pipeName"].ConvertToString();      
+
+            organizeList.Add(new OrganizeClient(delivery, config)); 
+        }
+    }
+}

dotnet/Sers/Sers.CL/Ipc/Sers.CL.Ipc.NamedPipe/Doc/Sers.CL.Ipc.NamedPipe - 额外命名管道/OrganizeServerBuilder.cs

@@ -0,0 +1,20 @@
+using System.Collections.Generic;
+using Newtonsoft.Json.Linq;
+using Sers.Core.CL.MessageOrganize;
+using Sers.Core.CL.MessageOrganize.DefaultOrganize;
+using Vit.Extensions;
+
+namespace Sers.CL.Ipc.NamedPipe
+{
+    public class OrganizeServerBuilder : IOrganizeServerBuilder
+    {
+        public void Build(List<IOrganizeServer> organizeList,JObject config)
+        {
+            var delivery = new DeliveryServer();
+ 
+            delivery.pipeName = config["pipeName"].ConvertToString();
+
+            organizeList.Add(new OrganizeServer(delivery, config)); 
+        }
+    }
+}

dotnet/Sers/Sers.CL/Ipc/Sers.CL.Ipc.NamedPipe/Doc/Sers.CL.Ipc.NamedPipe - 额外命名管道/Properties/PublishProfiles/FolderProfile.pubxml

@@ -0,0 +1,13 @@
+<?xml version="1.0" encoding="utf-8"?>
+<!--
+https://go.microsoft.com/fwlink/?LinkID=208121. 
+-->
+<Project ToolsVersion="4.0" xmlns="http://schemas.microsoft.com/developer/msbuild/2003">
+  <PropertyGroup>
+    <PublishProtocol>FileSystem</PublishProtocol>
+    <Configuration>Release</Configuration>
+    <Platform>Any CPU</Platform>
+    <TargetFramework>netstandard2.0</TargetFramework>
+    <PublishDir>..\..\..\..\Doc\Publish\nuget</PublishDir>
+  </PropertyGroup>
+</Project>

dotnet/Sers/Sers.CL/Ipc/Sers.CL.Ipc.NamedPipe/Doc/Sers.CL.Ipc.NamedPipe - 额外命名管道/Sers.CL.Ipc.NamedPipe.csproj

@@ -0,0 +1,26 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <TargetFramework>netstandard2.0</TargetFramework>
+    <Version>2.1.1.348</Version>
+    <Description>https://github.com/sersms/Sers/tree/2.1.1/release</Description>
+  </PropertyGroup>
+
+
+  <PropertyGroup>
+    <DocumentationFile>bin\Debug\netstandard2.0\Sers.CL.Ipc.NamedPipe.xml</DocumentationFile>
+  </PropertyGroup>
+
+
+  <ItemGroup>
+    <Compile Remove="Doc\**" />
+    <EmbeddedResource Remove="Doc\**" />
+    <None Remove="Doc\**" />
+  </ItemGroup>
+ 
+
+  <ItemGroup>
+    <ProjectReference Include="..\..\..\Sers.Core\Sers.Core\Sers.Core.csproj" />
+  </ItemGroup>
+
+</Project>

dotnet/Sers/Sers.CL/Ipc/Sers.CL.Ipc.NamedPipe/OrganizeClientBuilder.cs

@@ -0,0 +1,21 @@
+using System.Collections.Generic;
+using Newtonsoft.Json.Linq;
+using Sers.Core.CL.MessageOrganize;
+using Sers.Core.CL.MessageOrganize.DefaultOrganize;
+using Vit.Extensions;
+
+namespace Sers.CL.Ipc.NamedPipe
+{
+    public class OrganizeClientBuilder : IOrganizeClientBuilder
+    {
+        public void Build(List<IOrganizeClient> organizeList, JObject config)
+        {
+            var delivery = new DeliveryClient();
+
+            delivery.serverName = config["serverName"].ConvertToString();
+            delivery.pipeName = config["pipeName"].ConvertToString();      
+
+            organizeList.Add(new OrganizeClient(delivery, config)); 
+        }
+    }
+}

dotnet/Sers/Sers.CL/Ipc/Sers.CL.Ipc.NamedPipe/OrganizeServerBuilder.cs

@@ -0,0 +1,20 @@
+using System.Collections.Generic;
+using Newtonsoft.Json.Linq;
+using Sers.Core.CL.MessageOrganize;
+using Sers.Core.CL.MessageOrganize.DefaultOrganize;
+using Vit.Extensions;
+
+namespace Sers.CL.Ipc.NamedPipe
+{
+    public class OrganizeServerBuilder : IOrganizeServerBuilder
+    {
+        public void Build(List<IOrganizeServer> organizeList,JObject config)
+        {
+            var delivery = new DeliveryServer();
+ 
+            delivery.pipeName = config["pipeName"].ConvertToString();
+
+            organizeList.Add(new OrganizeServer(delivery, config)); 
+        }
+    }
+}

dotnet/Sers/Sers.CL/Ipc/Sers.CL.Ipc.NamedPipe/Properties/PublishProfiles/FolderProfile.pubxml

@@ -0,0 +1,13 @@
+<?xml version="1.0" encoding="utf-8"?>
+<!--
+https://go.microsoft.com/fwlink/?LinkID=208121. 
+-->
+<Project ToolsVersion="4.0" xmlns="http://schemas.microsoft.com/developer/msbuild/2003">
+  <PropertyGroup>
+    <PublishProtocol>FileSystem</PublishProtocol>
+    <Configuration>Release</Configuration>
+    <Platform>Any CPU</Platform>
+    <TargetFramework>netstandard2.0</TargetFramework>
+    <PublishDir>..\..\..\..\Doc\Publish\nuget</PublishDir>
+  </PropertyGroup>
+</Project>

dotnet/Sers/Sers.CL/Ipc/Sers.CL.Ipc.NamedPipe/Sers.CL.Ipc.NamedPipe.csproj

@@ -0,0 +1,26 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <TargetFramework>netstandard2.0</TargetFramework>
+    <Version>2.1.1.348</Version>
+    <Description>https://github.com/sersms/Sers/tree/2.1.1/release</Description>
+  </PropertyGroup>
+
+
+  <PropertyGroup>
+    <DocumentationFile>bin\Debug\netstandard2.0\Sers.CL.Ipc.NamedPipe.xml</DocumentationFile>
+  </PropertyGroup>
+
+
+  <ItemGroup>
+    <Compile Remove="Doc\**" />
+    <EmbeddedResource Remove="Doc\**" />
+    <None Remove="Doc\**" />
+  </ItemGroup>
+ 
+
+  <ItemGroup>
+    <ProjectReference Include="..\..\..\Sers.Core\Sers.Core\Sers.Core.csproj" />
+  </ItemGroup>
+
+</Project>

dotnet/Sers/Sers.CL/Test/CommunicationManage/Client/CLClient.csproj

@@ -16,6 +16,7 @@
   </ItemGroup>
 
   <ItemGroup>
+    <ProjectReference Include="..\..\..\Ipc\Sers.CL.Ipc.NamedPipe\Sers.CL.Ipc.NamedPipe.csproj" />
     <ProjectReference Include="..\..\..\Ipc\Sers.CL.Ipc.SharedMemory\Sers.CL.Ipc.SharedMemory.csproj" />
     <ProjectReference Include="..\..\..\Socket\Sers.CL.Socket.Iocp\Sers.CL.Socket.Iocp.csproj" />
     <ProjectReference Include="..\..\..\Socket\Sers.CL.Socket.ThreadWait\Sers.CL.Socket.ThreadWait.csproj" />

dotnet/Sers/Sers.CL/Test/CommunicationManage/Client/ProgramQps.cs

@@ -48,7 +48,9 @@ namespace CLClient
 
             cm.Conn_OnDisconnected = (conn) =>
             {
-                Logger.Info("Conn_OnDisconnected");
+                Logger.Info("Conn_OnDisconnected");       
+
+                Sers.Core.Module.App.SersApplication.OnStop();
             };
 
             cm.conn_OnGetRequest = (conn,sender,request,callback)=> 

dotnet/Sers/Sers.CL/Test/CommunicationManage/Client/appsettings.json

@@ -89,7 +89,7 @@
           /* 在此Assembly中查找Builder */
           "assemblyFile": "Sers.CL.Zmq.FullDuplex.dll",
           /* the class of Builder in assemblyFile  */
-          "className": "Sers.CL.Zmq.FullDuplex.OrganizeClientBuilder",
+          //"className": "Sers.CL.Zmq.FullDuplex.OrganizeClientBuilder",
 
 
           /* (x.2) config */
@@ -109,6 +109,21 @@
           /* (x.2) config */
           /* 共享内存名称。例如： "ipcTest" */
           "name": "ipcForStationDemo"
+        },
+        {
+          // Ipc.SharedMemory
+          /* (x.1) type - Ipc.SharedMemory */
+          /* 在此Assembly中查找Builder */
+          "assemblyFile": "Sers.CL.Ipc.NamedPipe.dll",
+          /* the class of Builder in assemblyFile  */
+          "className": "Sers.CL.Ipc.NamedPipe.OrganizeClientBuilder",
+
+
+          /* (x.2) config */
+          /* 共享内存名称。例如： "ipcTest" */
+          "serverName": ".",
+          /* 共享内存名称。例如： "ipcTest" */
+          "pipeName": "ipcForStationDemo"
         }
       ]
     }
@@ -119,7 +134,7 @@
 
   "PressureTest": {
     "clientCount": 1,
-    "requestThreadCount": 32,
+    "requestThreadCount": 1,
     "messageThreadCount": 0,
     "msgLen": 1
   }

dotnet/Sers/Sers.CL/Test/CommunicationManage/Server/CLServer.csproj

@@ -19,6 +19,7 @@
   </ItemGroup>
 
   <ItemGroup>
+    <ProjectReference Include="..\..\..\Ipc\Sers.CL.Ipc.NamedPipe\Sers.CL.Ipc.NamedPipe.csproj" />
     <ProjectReference Include="..\..\..\Ipc\Sers.CL.Ipc.SharedMemory\Sers.CL.Ipc.SharedMemory.csproj" />
     <ProjectReference Include="..\..\..\Socket\Sers.CL.Socket.Iocp\Sers.CL.Socket.Iocp.csproj" />
     <ProjectReference Include="..\..\..\Socket\Sers.CL.Socket.ThreadWait\Sers.CL.Socket.ThreadWait.csproj" />

dotnet/Sers/Sers.CL/Test/CommunicationManage/Server/Program.cs

@@ -49,8 +49,6 @@ namespace CLServer
 
             cm.Start();
 
-            Console.WriteLine("Hello World!");
-
             while (true)
             {                 
                 Thread.Sleep(5000);

dotnet/Sers/Sers.CL/Test/CommunicationManage/Server/appsettings.json

@@ -90,7 +90,7 @@
           /* 在此Assembly中查找Builder */
           "assemblyFile": "Sers.CL.Zmq.FullDuplex.dll",
           /* the class of Builder in assemblyFile  */
-          "className": "Sers.CL.Zmq.FullDuplex.OrganizeServerBuilder",
+          //"className": "Sers.CL.Zmq.FullDuplex.OrganizeServerBuilder",
 
 
           /* (x.2) config */
@@ -114,6 +114,20 @@
           "nodeCount": 128,
           /* 共享内存节点大小。例如： 2048 */
           "nodeBufferSize": 102400
+        },
+        {
+          //Ipc.SharedMemory
+          /* (x.1) type - Ipc.SharedMemory */
+          /* 在此Assembly中查找Builder */
+          "assemblyFile": "Sers.CL.Ipc.NamedPipe.dll",
+          /* the class of Builder in assemblyFile  */
+          "className": "Sers.CL.Ipc.NamedPipe.OrganizeServerBuilder",
+
+
+          /* (x.2) config */
+          /* 共享内存名称。例如： "ipcTest" */
+          "pipeName": "ipcForStationDemo"
+
         }
       ]
     }

dotnet/Sers/Sers.CL/Test/MessageDelivery/DeliveryTest/DeliveryTest.csproj

@@ -13,6 +13,7 @@
   </ItemGroup>
 
   <ItemGroup>
+    <ProjectReference Include="..\..\..\Ipc\Sers.CL.Ipc.NamedPipe\Sers.CL.Ipc.NamedPipe.csproj" />
     <ProjectReference Include="..\..\..\Ipc\Sers.CL.Ipc.SharedMemory\Sers.CL.Ipc.SharedMemory.csproj" />
     <ProjectReference Include="..\..\..\Socket\Sers.CL.Socket.Iocp\Sers.CL.Socket.Iocp.csproj" />
     <ProjectReference Include="..\..\..\Socket\Sers.CL.Socket.ThreadWait\Sers.CL.Socket.ThreadWait.csproj" />

dotnet/Sers/Sers.CL/Test/MessageDelivery/DeliveryTest/Program.cs

@@ -13,8 +13,6 @@ namespace DeliveryTest
         {
             Logger.OnLog = (level, msg) => { Console.Write("[" + level + "]" + msg); };
 
-
-
             StartServer();
             StartClient();
 
@@ -35,8 +33,9 @@ namespace DeliveryTest
             //var server = new Sers.CL.WebSocket.DeliveryServer();
             //var server = new Sers.CL.ClrZmq.ThreadWait.DeliveryServer();
             //var server = new Sers.CL.Ipc.SharedMemory.DeliveryServer();
-            var server = new Sers.CL.Zmq.FullDuplex.DeliveryServer();
- 
+            //var server = new Sers.CL.Zmq.FullDuplex.DeliveryServer();
+            var server = new Sers.CL.Ipc.NamedPipe.DeliveryServer();
+
             server.Conn_OnConnected = (conn) => 
             {
                 conn.OnGetFrame = (conn_, data) =>
@@ -48,7 +47,6 @@ namespace DeliveryTest
                     conn_.SendFrameAsync(byteData);
                 };
             };
-
            
 
             server.Start();
@@ -63,7 +61,8 @@ namespace DeliveryTest
             //var client = new Sers.CL.WebSocket.DeliveryClient();
             //var client = new Sers.CL.ClrZmq.ThreadWait.DeliveryClient();
             //var client = new Sers.CL.Ipc.SharedMemory.DeliveryClient();
-            var client = new Sers.CL.Zmq.FullDuplex.DeliveryClient();
+            //var client = new Sers.CL.Zmq.FullDuplex.DeliveryClient();
+            var client = new Sers.CL.Ipc.NamedPipe.DeliveryClient();
 
             client.Conn_OnGetFrame = (conn, data) =>
             {

dotnet/Sers/Sers.Core/Sers.Core/CL/MessageOrganize/DefaultOrganize/OrganizeServer.cs

@@ -81,6 +81,12 @@ namespace Sers.Core.CL.MessageOrganize.DefaultOrganize
 
         public bool Start()
         {
+            lock (this)
+            {
+                if (isRunning) return true;
+                isRunning = true;
+            }
+
             requestAdaptor.Start();
             if (!delivery.Start())
             {
@@ -89,10 +95,16 @@ namespace Sers.Core.CL.MessageOrganize.DefaultOrganize
             }
             return true;
         }
-
+        bool isRunning = false;
 
         public void Stop()
-        {           
+        {
+            lock (this)
+            {
+                if (!isRunning) return;
+                isRunning = false;
+            }
+
             requestAdaptor.Stop();
             delivery.Stop();
         }