1.消除了了网络层的MemoryBuffer的GC

2.单线程不再走线程之间的Operator队列，而是直接处理，提升了单线程的性能

Unity/Assets/Scripts/Codes/Hotfix/Share/Module/Move/MoveComponentSystem.cs

@@ -28,7 +28,7 @@ namespace ET
         {
             protected override void Destroy(MoveComponent self)
             {
-                self.MoveFinish(true);
+                self.MoveFinish(false);
             }
         }
 

Unity/Assets/Scripts/Core/Module/Network/AService.cs

@@ -10,11 +10,10 @@ namespace ET
         
         public ServiceType ServiceType { get; protected set; }
         
-        private (object Message, MemoryStream MemoryStream) lastMessageInfo;
+        private (object Message, MemoryBuffer MemoryStream) lastMessageInfo;
         
-        // 缓存上一个发送的消息，这样广播消息的时候省掉多次序列化,这样有个另外的问题,客户端就不能保存发送的消息来减少gc，
-        // 不过这个问题并不大，因为客户端发送的消息是比较少的，如果实在需要，也可以修改这个方法，把outer的消息过滤掉。
-        protected MemoryStream GetMemoryStream(object message)
+        // 缓存上一个发送的消息，这样广播消息的时候省掉多次序列化,这样有个另外的问题,客户端就不能保存发送的消息来减少gc
+        public MemoryBuffer Fetch(object message)
         {
             if (object.ReferenceEquals(lastMessageInfo.Message, message))
             {
@@ -22,10 +21,31 @@ namespace ET
                 return lastMessageInfo.MemoryStream;
             }
 
-            (ushort _, MemoryStream stream) = MessageSerializeHelper.MessageToStream(message);
+            MemoryBuffer stream = NetServices.Instance.Fetch();
+            MessageSerializeHelper.MessageToStream(stream, message);
             this.lastMessageInfo = (message, stream);
             return stream;
         }
+
+        private MemoryBuffer lastRecyleMessage;
+        
+        // Recycle不能直接回到池中，因为有可能广播消息的时候，发送的是同一个MemoryBuff
+        public void Recycle(MemoryBuffer memoryStream)
+        {
+            if (this.lastRecyleMessage == null)
+            {
+                this.lastRecyleMessage = memoryStream;
+                return;
+            }
+
+            if (ReferenceEquals(this.lastRecyleMessage, memoryStream))
+            {
+                return;
+            }
+            
+            NetServices.Instance.Recycle(this.lastRecyleMessage);
+            this.lastRecyleMessage = memoryStream;
+        }
         
         public virtual void Dispose()
         {

Unity/Assets/Scripts/Core/Module/Network/KChannel.cs

@@ -12,7 +12,7 @@ namespace ET
 	public struct KcpWaitPacket
 	{
 		public long ActorId;
-		public MemoryStream MemoryStream;
+		public MemoryBuffer MemoryStream;
 	}
 	
 	public class KChannel : AChannel
@@ -387,7 +387,7 @@ namespace ET
 			{
 				return;
 			}
-			MemoryStream memoryStream = kcpWaitPacket.MemoryStream;
+			MemoryBuffer memoryStream = kcpWaitPacket.MemoryStream;
 			
 			switch (this.Service.ServiceType)
 			{
@@ -432,11 +432,16 @@ namespace ET
 			}
 
 			this.Service.AddToUpdate(0, this.Id);
+			
+			// 回收MemoryBuffer，减少GC
+			this.Service.Recycle(memoryStream);
 		}
 		
-		public void Send(long actorId, MemoryStream stream)
+		public void Send(long actorId, object message)
 		{
-			KcpWaitPacket kcpWaitPacket = new KcpWaitPacket() { ActorId = actorId, MemoryStream = stream };
+			MemoryBuffer stream = this.Service.Fetch(message);
+			MessageSerializeHelper.MessageToStream(stream, message);
+			KcpWaitPacket kcpWaitPacket = new() { ActorId = actorId, MemoryStream = stream };
 			if (!this.IsConnected)
 			{
 				this.sendBuffer.Enqueue(kcpWaitPacket);
@@ -469,6 +474,7 @@ namespace ET
 				this.OnError(ErrorCore.ERR_KcpWaitSendSizeTooLarge);
 				return;
 			}
+
 			this.KcpSend(kcpWaitPacket);
 		}
 		

Unity/Assets/Scripts/Core/Module/Network/KService.cs

@@ -527,8 +527,7 @@ namespace ET
                 return;
             }
             
-            MemoryStream memoryStream = this.GetMemoryStream(message);
-            channel.Send(actorId, memoryStream);
+            channel.Send(actorId, message);
         }
 
         public override void Update()

Unity/Assets/Scripts/Core/Module/Network/MessageSerializeHelper.cs

@@ -5,25 +5,9 @@ namespace ET
 {
     public static class MessageSerializeHelper
     {
-        private static MemoryBuffer GetStream(int count = 0)
-        {
-            MemoryBuffer stream;
-            if (count > 0)
-            {
-                stream = new MemoryBuffer(count);
-            }
-            else
-            {
-                stream = new MemoryBuffer();
-            }
-
-            return stream;
-        }
-        
-        public static (ushort, MemoryBuffer) MessageToStream(object message)
+        public static ushort MessageToStream(MemoryBuffer stream, object message)
         {
             int headOffset = Packet.ActorIdLength;
-            MemoryBuffer stream = GetStream(headOffset + Packet.OpcodeLength);
 
             ushort opcode = NetServices.Instance.GetOpcode(message.GetType());
             
@@ -35,7 +19,7 @@ namespace ET
             SerializeHelper.Serialize(message, stream);
             
             stream.Seek(0, SeekOrigin.Begin);
-            return (opcode, stream);
+            return opcode;
         }
     }
 }

Unity/Assets/Scripts/Core/Module/Network/NetServices.cs

@@ -1,4 +1,6 @@
-using System;
+//#undef SINGLE_THREAD
+
+using System;
 using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.IO;
@@ -6,6 +8,8 @@ using System.Net;
 using System.Threading;
 using System.Threading.Tasks;
 
+
+
 namespace ET
 {
     public enum NetworkProtocol
@@ -14,7 +18,7 @@ namespace ET
         KCP,
         Websocket,
     }
-    
+
     public enum NetOp: byte
     {
         AddService = 1,
@@ -28,7 +32,7 @@ namespace ET
         GetChannelConn = 10,
         ChangeAddress = 11,
     }
-    
+
     public struct NetOperator
     {
         public NetOp Op; // 操作码
@@ -40,8 +44,10 @@ namespace ET
 
     public class NetServices: Singleton<NetServices>, ISingletonUpdate
     {
+#if !SINGLE_THREAD
         private readonly ConcurrentQueue<NetOperator> netThreadOperators = new ConcurrentQueue<NetOperator>();
         private readonly ConcurrentQueue<NetOperator> mainThreadOperators = new ConcurrentQueue<NetOperator>();
+#endif
 
         public NetServices()
         {
@@ -62,7 +68,7 @@ namespace ET
 
                 this.typeOpcode.Add(type, messageAttribute.Opcode);
             }
-            
+
 #if !SINGLE_THREAD
             // 网络线程
             this.thread = new Thread(this.NetThreadUpdate);
@@ -72,15 +78,14 @@ namespace ET
 
         public void Destroy()
         {
-            
 #if !SINGLE_THREAD
-            this.isStop = true;            
+            this.isStop = true;
             this.thread.Join(1000);
 #endif
         }
 
-#region 线程安全
-        
+        #region 线程安全
+
         // 初始化后不变，所以主线程，网络线程都可以读
         private readonly DoubleMap<Type, ushort> typeOpcode = new DoubleMap<Type, ushort>();
 
@@ -94,139 +99,200 @@ namespace ET
             return this.typeOpcode.GetKeyByValue(opcode);
         }
 
-#endregion
+        #endregion
+
+        #region 主线程
+
+        private readonly Dictionary<int, Action<long, IPEndPoint>> acceptCallback = new();
+        private readonly Dictionary<int, Action<long, long, object>> readCallback = new();
+        private readonly Dictionary<int, Action<long, int>> errorCallback = new();
 
-        
-        
-#region 主线程
-        
-        private readonly Dictionary<int, Action<long, IPEndPoint>> acceptCallback = new Dictionary<int, Action<long, IPEndPoint>>();
-        private readonly Dictionary<int, Action<long, long, object>> readCallback = new Dictionary<int, Action<long, long, object>>();
-        private readonly Dictionary<int, Action<long, int>> errorCallback = new Dictionary<int, Action<long, int>>();
-        
         private int serviceIdGenerator;
 
         public async Task<(uint, uint)> GetChannelConn(int serviceId, long channelId)
         {
             TaskCompletionSource<(uint, uint)> tcs = new TaskCompletionSource<(uint, uint)>();
-            NetOperator netOperator = new NetOperator() { Op = NetOp.GetChannelConn, ServiceId = serviceId, ChannelId = channelId, Object = tcs};
-            this.netThreadOperators.Enqueue(netOperator);
+            NetOperator netOperator = new NetOperator() { Op = NetOp.GetChannelConn, ServiceId = serviceId, ChannelId = channelId, Object = tcs };
+            ToNetThread(ref netOperator);
             return await tcs.Task;
         }
 
-        public void ChangeAddress(int serviceId, long channelId, IPEndPoint ipEndPoint)
+        private void ToNetThread(ref NetOperator netOperator)
         {
-            NetOperator netOperator = new NetOperator() { Op = NetOp.ChangeAddress, ServiceId = serviceId, ChannelId = channelId, Object = ipEndPoint};
+#if !SINGLE_THREAD
             this.netThreadOperators.Enqueue(netOperator);
+#else
+            NetThreadExecute(ref netOperator);
+#endif
+        }
+
+        private void ToMainThread(ref NetOperator netOperator)
+        {
+#if !SINGLE_THREAD
+            this.mainThreadOperators.Enqueue(netOperator);
+#else
+            MainThreadExecute(ref netOperator);
+#endif
         }
-        
+
+        public void ChangeAddress(int serviceId, long channelId, IPEndPoint ipEndPoint)
+        {
+            NetOperator netOperator =
+                    new NetOperator() { Op = NetOp.ChangeAddress, ServiceId = serviceId, ChannelId = channelId, Object = ipEndPoint };
+            ToNetThread(ref netOperator);
+        }
+
         public void SendMessage(int serviceId, long channelId, long actorId, object message)
         {
-            NetOperator netOperator = new NetOperator() { Op = NetOp.SendMessage, ServiceId = serviceId, ChannelId = channelId, ActorId = actorId, Object = message };
-            this.netThreadOperators.Enqueue(netOperator);
+            NetOperator netOperator = new NetOperator()
+            {
+                Op = NetOp.SendMessage,
+                ServiceId = serviceId,
+                ChannelId = channelId,
+                ActorId = actorId,
+                Object = message
+            };
+            ToNetThread(ref netOperator);
         }
 
         public int AddService(AService aService)
         {
             aService.Id = ++this.serviceIdGenerator;
             NetOperator netOperator = new NetOperator() { Op = NetOp.AddService, ServiceId = aService.Id, ChannelId = 0, Object = aService };
-            this.netThreadOperators.Enqueue(netOperator);
+            ToNetThread(ref netOperator);
             return aService.Id;
         }
-        
+
         public void RemoveService(int serviceId)
         {
             NetOperator netOperator = new NetOperator() { Op = NetOp.RemoveService, ServiceId = serviceId };
-            this.netThreadOperators.Enqueue(netOperator);
+            ToNetThread(ref netOperator);
         }
-        
+
         public void RemoveChannel(int serviceId, long channelId, int error)
         {
-            NetOperator netOperator = new NetOperator() { Op = NetOp.RemoveChannel, ServiceId = serviceId, ChannelId = channelId, ActorId = error};
-            this.netThreadOperators.Enqueue(netOperator);
+            NetOperator netOperator = new NetOperator() { Op = NetOp.RemoveChannel, ServiceId = serviceId, ChannelId = channelId, ActorId = error };
+            ToNetThread(ref netOperator);
         }
 
         public void CreateChannel(int serviceId, long channelId, IPEndPoint address)
         {
-            NetOperator netOperator = new NetOperator() { Op = NetOp.CreateChannel, ServiceId = serviceId, ChannelId = channelId, Object = address};
-            this.netThreadOperators.Enqueue(netOperator);
+            NetOperator netOperator = new NetOperator() { Op = NetOp.CreateChannel, ServiceId = serviceId, ChannelId = channelId, Object = address };
+            ToNetThread(ref netOperator);
         }
 
         public void RegisterAcceptCallback(int serviceId, Action<long, IPEndPoint> action)
         {
             this.acceptCallback.Add(serviceId, action);
         }
-        
+
         public void RegisterReadCallback(int serviceId, Action<long, long, object> action)
         {
             this.readCallback.Add(serviceId, action);
         }
-        
+
         public void RegisterErrorCallback(int serviceId, Action<long, int> action)
         {
             this.errorCallback.Add(serviceId, action);
         }
-        
-        private void UpdateInMainThread()
+
+        private void MainThreadExecute(ref NetOperator op)
         {
-            while (true)
+            try
             {
-                if (!this.mainThreadOperators.TryDequeue(out NetOperator op))
+                switch (op.Op)
                 {
-                    return;
-                }
-
-                try
-                {
-                    switch (op.Op)
+                    case NetOp.OnAccept:
                     {
-                        case NetOp.OnAccept:
+                        if (!this.acceptCallback.TryGetValue(op.ServiceId, out var action))
                         {
-                            if (!this.acceptCallback.TryGetValue(op.ServiceId, out var action))
-                            {
-                                return;
-                            }
-                            action.Invoke(op.ChannelId, op.Object as IPEndPoint);
-                            break;
+                            return;
                         }
-                        case NetOp.OnRead:
+
+                        action.Invoke(op.ChannelId, op.Object as IPEndPoint);
+                        break;
+                    }
+                    case NetOp.OnRead:
+                    {
+                        if (!this.readCallback.TryGetValue(op.ServiceId, out var action))
                         {
-                            if (!this.readCallback.TryGetValue(op.ServiceId, out var action))
-                            {
-                                return;
-                            }
-                            action.Invoke(op.ChannelId, op.ActorId, op.Object);
-                            break;
+                            return;
                         }
-                        case NetOp.OnError:
+
+                        action.Invoke(op.ChannelId, op.ActorId, op.Object);
+                        break;
+                    }
+                    case NetOp.OnError:
+                    {
+                        if (!this.errorCallback.TryGetValue(op.ServiceId, out var action))
                         {
-                            if (!this.errorCallback.TryGetValue(op.ServiceId, out var action))
-                            {
-                                return;
-                            }
-                            
-                            action.Invoke(op.ChannelId, (int)op.ActorId);
-                            break;
+                            return;
                         }
-                        default:
-                            throw new Exception($"not found net operator: {op.Op}");
+
+                        action.Invoke(op.ChannelId, (int) op.ActorId);
+                        break;
                     }
+                    default:
+                        throw new Exception($"not found net operator: {op.Op}");
                 }
-                catch (Exception e)
+            }
+            catch (Exception e)
+            {
+                Log.Error(e);
+            }
+        }
+
+        private void UpdateInMainThread()
+        {
+#if !SINGLE_THREAD
+            while (true)
+            {
+                if (!this.mainThreadOperators.TryDequeue(out NetOperator op))
                 {
-                    Log.Error(e);
+                    return;
                 }
+
+                MainThreadExecute(ref op);
             }
+#endif
         }
 
-#endregion
+        #endregion
+
+        #region 网络线程
 
-#region 网络线程
-        
         private readonly Dictionary<int, AService> services = new Dictionary<int, AService>();
         private readonly Queue<int> queue = new Queue<int>();
-        
-        
+
+        private readonly Queue<MemoryBuffer> pool = new Queue<MemoryBuffer>();
+
+        public MemoryBuffer Fetch()
+        {
+            if (this.pool.Count > 0)
+            {
+                return this.pool.Dequeue();
+            }
+
+            return new MemoryBuffer();
+        }
+
+        public void Recycle(MemoryBuffer memoryBuffer)
+        {
+            if (memoryBuffer.Capacity > 128) // 太大的不回收，GC
+            {
+                return;
+            }
+
+            if (this.pool.Count > 1000)
+            {
+                return;
+            }
+
+            memoryBuffer.SetLength(0);
+            memoryBuffer.Seek(0, SeekOrigin.Begin);
+            this.pool.Enqueue(memoryBuffer);
+        }
+
         private void Add(AService aService)
         {
             this.services[aService.Id] = aService;
@@ -239,7 +305,7 @@ namespace ET
             this.services.TryGetValue(id, out aService);
             return aService;
         }
-        
+
         private void Remove(int id)
         {
             if (this.services.Remove(id, out AService service))
@@ -249,10 +315,10 @@ namespace ET
         }
 
 #if !SINGLE_THREAD
-        
+
         private bool isStop;
         private readonly Thread thread;
-        
+
         // 网络线程Update
         private void NetThreadUpdate()
         {
@@ -261,99 +327,114 @@ namespace ET
                 this.UpdateInNetThread();
                 Thread.Sleep(1);
             }
+
+            // 停止的时候再执行一帧，把队列中的消息处理完成
+            this.UpdateInNetThread();
         }
 #endif
 
-        private void RunNetThreadOperator()
+        private void NetThreadExecute(ref NetOperator op)
         {
-            while (true)
+            try
             {
-                if (!this.netThreadOperators.TryDequeue(out NetOperator op))
+                switch (op.Op)
                 {
-                    return;
-                }
-
-                try
-                {
-                    switch (op.Op)
+                    case NetOp.AddService:
+                    {
+                        this.Add(op.Object as AService);
+                        break;
+                    }
+                    case NetOp.RemoveService:
+                    {
+                        this.Remove(op.ServiceId);
+                        break;
+                    }
+                    case NetOp.CreateChannel:
                     {
-                        case NetOp.AddService:
+                        AService service = this.Get(op.ServiceId);
+                        if (service != null)
                         {
-                            this.Add(op.Object as AService);
-                            break;
+                            service.Create(op.ChannelId, op.Object as IPEndPoint);
                         }
-                        case NetOp.RemoveService:
+
+                        break;
+                    }
+                    case NetOp.RemoveChannel:
+                    {
+                        AService service = this.Get(op.ServiceId);
+                        if (service != null)
                         {
-                            this.Remove(op.ServiceId);
-                            break;
+                            service.Remove(op.ChannelId, (int) op.ActorId);
                         }
-                        case NetOp.CreateChannel:
+
+                        break;
+                    }
+                    case NetOp.SendMessage:
+                    {
+                        AService service = this.Get(op.ServiceId);
+                        if (service != null)
                         {
-                            AService service = this.Get(op.ServiceId);
-                            if (service != null)
-                            {
-                                service.Create(op.ChannelId, op.Object as IPEndPoint);
-                            }
-                            break;
+                            service.Send(op.ChannelId, op.ActorId, op.Object);
                         }
-                        case NetOp.RemoveChannel:
+
+                        break;
+                    }
+                    case NetOp.GetChannelConn:
+                    {
+                        var tcs = op.Object as TaskCompletionSource<ValueTuple<uint, uint>>;
+                        try
                         {
                             AService service = this.Get(op.ServiceId);
-                            if (service != null)
+                            if (service == null)
                             {
-                                service.Remove(op.ChannelId, (int)op.ActorId);
+                                break;
                             }
-                            break;
+
+                            tcs.SetResult(service.GetChannelConn(op.ChannelId));
                         }
-                        case NetOp.SendMessage:
+                        catch (Exception e)
                         {
-                            AService service = this.Get(op.ServiceId);
-                            if (service != null)
-                            {
-                                service.Send(op.ChannelId, op.ActorId, op.Object);
-                            }
-                            break;
+                            tcs.SetException(e);
                         }
-                        case NetOp.GetChannelConn:
-                        {
-                            var tcs = op.Object as TaskCompletionSource<ValueTuple<uint, uint>>;
-                            try
-                            {
-                                AService service = this.Get(op.ServiceId);
-                                if (service == null)
-                                {
-                                    break;
-                                }
 
-                                tcs.SetResult(service.GetChannelConn(op.ChannelId));
-                            }
-                            catch (Exception e)
-                            {
-                                tcs.SetException(e);
-                            }
-                            break;
-                        }
-                        case NetOp.ChangeAddress:
+                        break;
+                    }
+                    case NetOp.ChangeAddress:
+                    {
+                        AService service = this.Get(op.ServiceId);
+                        if (service == null)
                         {
-                            AService service = this.Get(op.ServiceId);
-                            if (service == null)
-                            {
-                                break;
-                            }
-                            service.ChangeAddress(op.ChannelId, op.Object as IPEndPoint);
                             break;
                         }
-                        default:
-                            throw new Exception($"not found net operator: {op.Op}");
+
+                        service.ChangeAddress(op.ChannelId, op.Object as IPEndPoint);
+                        break;
                     }
+                    default:
+                        throw new Exception($"not found net operator: {op.Op}");
                 }
-                catch (Exception e)
+            }
+            catch (Exception e)
+            {
+                Log.Error(e);
+            }
+        }
+
+        private void RunNetThreadOperator()
+        {
+#if !SINGLE_THREAD
+            while (true)
+            {
+                if (!this.netThreadOperators.TryDequeue(out NetOperator op))
                 {
-                    Log.Error(e);
+                    return;
                 }
+                
+                NetThreadExecute(ref op);
             }
+#endif
         }
-        
+
         private void UpdateInNetThread()
         {
             int count = this.queue.Count;
@@ -364,34 +445,42 @@ namespace ET
                 {
                     continue;
                 }
+
                 this.queue.Enqueue(serviceId);
                 service.Update();
             }
-            
+
             this.RunNetThreadOperator();
         }
 
         public void OnAccept(int serviceId, long channelId, IPEndPoint ipEndPoint)
         {
             NetOperator netOperator = new NetOperator() { Op = NetOp.OnAccept, ServiceId = serviceId, ChannelId = channelId, Object = ipEndPoint };
-            this.mainThreadOperators.Enqueue(netOperator);
+            ToMainThread(ref netOperator);
         }
 
         public void OnRead(int serviceId, long channelId, long actorId, object message)
         {
-            NetOperator netOperator = new NetOperator() { Op = NetOp.OnRead, ServiceId = serviceId, ChannelId = channelId, ActorId = actorId, Object = message };
-            this.mainThreadOperators.Enqueue(netOperator);
+            NetOperator netOperator = new NetOperator()
+            {
+                Op = NetOp.OnRead,
+                ServiceId = serviceId,
+                ChannelId = channelId,
+                ActorId = actorId,
+                Object = message
+            };
+            ToMainThread(ref netOperator);
         }
 
         public void OnError(int serviceId, long channelId, int error)
         {
             NetOperator netOperator = new NetOperator() { Op = NetOp.OnError, ServiceId = serviceId, ChannelId = channelId, ActorId = error };
-            this.mainThreadOperators.Enqueue(netOperator);
+            ToMainThread(ref netOperator);
         }
 
-#endregion
+        #endregion
 
-#region 主线程kcp id生成
+        #region 主线程kcp id生成
 
         // 这个因为是NetClientComponent中使用，不会与Accept冲突
         public uint CreateConnectChannelId()
@@ -399,18 +488,19 @@ namespace ET
             return RandomGenerator.RandUInt32();
         }
 
-#endregion
+        #endregion
 
-#region 网络线程kcp id生成
+        #region 网络线程kcp id生成
 
         // 防止与内网进程号的ChannelId冲突，所以设置为一个大的随机数
         private uint acceptIdGenerator = uint.MaxValue;
+
         public uint CreateAcceptChannelId()
         {
             return --this.acceptIdGenerator;
         }
 
-#endregion
+        #endregion
 
         public void Update()
         {

Unity/Assets/Scripts/Core/Module/Network/TChannel.cs

@@ -90,7 +90,7 @@ namespace ET
 			this.socket = null;
 		}
 
-		public void Send(long actorId, MemoryStream stream)
+		public void Send(long actorId, MemoryBuffer stream)
 		{
 			if (this.IsDisposed)
 			{
@@ -128,6 +128,8 @@ namespace ET
 				}
 			}
 			
+			this.Service.Recycle(stream);
+			
 			if (!this.isSending)
 			{
 				//this.StartSend();

Unity/Assets/Scripts/Core/Module/Network/TService.cs

@@ -171,7 +171,7 @@ namespace ET
 					NetServices.Instance.OnError(this.Id, channelId, ErrorCore.ERR_SendMessageNotFoundTChannel);
 					return;
 				}
-				MemoryStream memoryStream = this.GetMemoryStream(message);
+				MemoryBuffer memoryStream = this.Fetch(message);
 				aChannel.Send(actorId, memoryStream);
 			}
 			catch (Exception e)

Unity/Assets/Scripts/Core/Module/Network/WService.cs

@@ -140,7 +140,7 @@ namespace ET
             {
                 return;
             }
-            MemoryStream memoryStream = this.GetMemoryStream(message);
+            MemoryStream memoryStream = this.Fetch(message);
             channel.Send(memoryStream);
         }