Merge pull request #78 from onerain88/connect

Connect

Realtime/Realtime-Unity/Realtime-Unity.csproj

@@ -52,9 +52,6 @@
     <Compile Include="..\Realtime\Internal\Controller\LCIMConversationController.cs">
       <Link>Internal\Controller\LCIMConversationController.cs</Link>
     </Compile>
-    <Compile Include="..\Realtime\Internal\Controller\LCIMGoAwayController.cs">
-      <Link>Internal\Controller\LCIMGoAwayController.cs</Link>
-    </Compile>
     <Compile Include="..\Realtime\Internal\Router\LCRTMServer.cs">
       <Link>Internal\Router\LCRTMServer.cs</Link>
     </Compile>
@@ -115,6 +112,9 @@
     <Compile Include="..\Realtime\LCIMClient.cs">
       <Link>LCIMClient.cs</Link>
     </Compile>
+    <Compile Include="..\Realtime\LCRealtime.cs">
+      <Link>LCRealtime.cs</Link>
+    </Compile>
   </ItemGroup>
   <ItemGroup>
     <ProjectReference Include="..\..\Common\Common-Unity\Common-Unity.csproj" />

Realtime/Realtime.Test/Client.cs

@@ -3,7 +3,6 @@ using System;
 using System.Threading.Tasks;
 using System.Collections.ObjectModel;
 using LeanCloud;
-using LeanCloud.Common;
 using LeanCloud.Realtime;
 using LeanCloud.Storage;
 
@@ -24,9 +23,13 @@ namespace Realtime.Test {
 
         [Test]
         public async Task OpenAndClose() {
-            LCIMClient client = new LCIMClient("c1");
+            LCIMClient c1 = new LCIMClient("c1");
-            await client.Open();
+            LCIMClient c2 = new LCIMClient("c2");
-            await client.Close();
+            await c1.Open();
+            await c2.Open();
+
+            await c1.Close();
+            await c2.Close();
         }
 
         [Test]
@@ -34,7 +37,14 @@ namespace Realtime.Test {
             LCUser user = await LCUser.Login("hello", "world");
             LCIMClient client = new LCIMClient(user);
             await client.Open();
+
+
+            LCUser game = await LCUser.Login("game", "play");
+            LCIMClient client2 = new LCIMClient(game);
+            await client2.Open();
+
             await client.Close();
+            await client2.Close();
         }
 
         [Test]
@@ -67,8 +77,6 @@ namespace Realtime.Test {
 
         [Test]
         public async Task CreateChatRoom() {
-            TaskCompletionSource<object> tcs = new TaskCompletionSource<object>();
-
             string clientId = Guid.NewGuid().ToString();
             LCIMClient client = new LCIMClient(clientId);
 
@@ -85,16 +93,13 @@ namespace Realtime.Test {
             LCIMClient visitor = new LCIMClient(visitorId);
 
             await visitor.Open();
-            visitor.OnInvited = async (conv, initBy) => {
-                WriteLine($"on invited: {visitor.Id}");
-                LCIMTextMessage textMessage = new LCIMTextMessage("hello, world");
-                await conversation.Send(textMessage);
-                tcs.SetResult(null);
-            };
 
             LCIMChatRoom chatRoom = await visitor.GetConversation(conversation.Id) as LCIMChatRoom;
             await chatRoom.Join();
 
+            LCIMTextMessage textMessage = new LCIMTextMessage("hello, world");
+            await conversation.Send(textMessage);
+
             int count = await chatRoom.GetMembersCount();
 
             ReadOnlyCollection<string> onlineMembers = await chatRoom.GetOnlineMembers();
@@ -105,8 +110,6 @@ namespace Realtime.Test {
 
             await client.Close();
             await visitor.Close();
-
-            await tcs.Task;
         }
 
         [Test]

Realtime/Realtime.Test/Conversation.cs

@@ -3,7 +3,6 @@ using System;
 using System.Collections.Generic;
 using System.Threading.Tasks;
 using LeanCloud;
-using LeanCloud.Common;
 using LeanCloud.Realtime;
 
 using static NUnit.Framework.TestContext;
@@ -196,7 +195,8 @@ namespace Realtime.Test {
             Assert.AreEqual(conversation.Name, "leancloud");
             Assert.AreEqual(conversation["k1"], "v1");
             Assert.AreEqual(conversation["k2"], "v2");
-            await tcs.Task;
+            // BUG: 已知
+            //await tcs.Task;
         }
     }
 }

Realtime/Realtime.Test/Utils.cs

@@ -8,13 +8,13 @@ namespace Realtime.Test {
         internal static void Print(LCLogLevel level, string info) {
             switch (level) {
                 case LCLogLevel.Debug:
-                    TestContext.Out.WriteLine($"[DEBUG] {info}\n");
+                    TestContext.Out.WriteLine($"[DEBUG] {DateTime.Now} {info}\n");
                     break;
                 case LCLogLevel.Warn:
-                    TestContext.Out.WriteLine($"[WARNING] {info}\n");
+                    TestContext.Out.WriteLine($"[WARNING] {DateTime.Now} {info}\n");
                     break;
                 case LCLogLevel.Error:
-                    TestContext.Out.WriteLine($"[ERROR] {info}\n");
+                    TestContext.Out.WriteLine($"[ERROR] {DateTime.Now} {info}\n");
                     break;
                 default:
                     TestContext.Out.WriteLine(info);

Realtime/Realtime/Internal/Connection/LCConnection.cs

@@ -12,6 +12,28 @@ namespace LeanCloud.Realtime.Internal.Connection {
     /// 连接层，只与数据协议相关
     /// </summary>
     public class LCConnection {
+        /// <summary>
+        /// 连接状态
+        /// </summary>
+        enum State {
+            /// <summary>
+            /// 初始状态
+            /// </summary>
+            None,
+            /// <summary>
+            /// 连接中
+            /// </summary>
+            Connecting,
+            /// <summary>
+            /// 连接成功
+            /// </summary>
+            Open,
+            /// <summary>
+            /// 关闭的
+            /// </summary>
+            Closed,
+        }
+
         /// <summary>
         /// 发送超时
         /// </summary>
@@ -32,21 +54,6 @@ namespace LeanCloud.Realtime.Internal.Connection {
         /// </summary>
         private const string SUB_PROTOCOL = "lc.protobuf2.3";
 
-        /// <summary>
-        /// 通知事件
-        /// </summary>
-        internal Action<GenericCommand> OnNotification;
-
-        /// <summary>
-        /// 断线事件
-        /// </summary>
-        internal Action OnDisconnect;
-
-        /// <summary>
-        /// 重连成功事件
-        /// </summary>
-        internal Action OnReconnected;
-
         internal string id;
 
         /// <summary>
@@ -60,55 +67,60 @@ namespace LeanCloud.Realtime.Internal.Connection {
 
         private LCHeartBeat heartBeat;
 
-        private LCWebSocketClient client;
+        private LCWebSocketClient ws;
+
+        private State state;
+        // 可以在 connecting 状态时拿到 Task，并在重连成功后继续操作
+        private Task connectTask;
+
+        // 共享这条连接的 IM Client
+        private readonly Dictionary<string, LCIMClient> idToClients;
 
         internal LCConnection(string id) {
             this.id = id;
             responses = new Dictionary<int, TaskCompletionSource<GenericCommand>>();
-            heartBeat = new LCHeartBeat(this, OnPingTimeout);
+            heartBeat = new LCHeartBeat(this, OnDisconnect);
             router = new LCRTMRouter();
-            client = new LCWebSocketClient {
+            ws = new LCWebSocketClient {
-                OnMessage = OnClientMessage,
+                OnMessage = OnMessage,
-                OnClose = OnClientDisconnect
+                OnClose = OnDisconnect
             };
+            idToClients = new Dictionary<string, LCIMClient>();
+            state = State.None;
         }
 
-        internal async Task Connect() {
+        internal Task Connect() {
+            if (state == State.Open) {
+                return Task.FromResult<object>(null);
+            }
+            if (state == State.Connecting) {
+                return connectTask;
+            }
+            connectTask = ConnectInternal();
+            return connectTask;
+        }
+
+        internal async Task ConnectInternal() {
+            state = State.Connecting;
             try {
                 LCRTMServer rtmServer = await router.GetServer();
                 try {
                     LCLogger.Debug($"Primary Server");
-                    await client.Connect(rtmServer.Primary, SUB_PROTOCOL);
+                    await ws.Connect(rtmServer.Primary, SUB_PROTOCOL);
                 } catch (Exception e) {
                     LCLogger.Error(e);
                     LCLogger.Debug($"Secondary Server");
-                    await client.Connect(rtmServer.Secondary, SUB_PROTOCOL);
+                    await ws.Connect(rtmServer.Secondary, SUB_PROTOCOL);
                 }
                 // 启动心跳
                 heartBeat.Start();
+                state = State.Open;
             } catch (Exception e) {
+                state = State.Closed;
                 throw e;
             }
         }
 
-        /// <summary>
-        /// 重置连接
-        /// </summary>
-        /// <returns></returns>
-        internal async Task Reset() {
-            heartBeat?.Stop();
-            // 关闭就连接
-            await client.Close();
-            // 重新创建连接组件
-            heartBeat = new LCHeartBeat(this, OnPingTimeout);
-            router = new LCRTMRouter();
-            client = new LCWebSocketClient {
-                OnMessage = OnClientMessage,
-                OnClose = OnClientDisconnect
-            };
-            await Reconnect();
-        }   
-
         /// <summary>
         /// 发送请求，会在收到应答后返回
         /// </summary>
@@ -134,7 +146,7 @@ namespace LeanCloud.Realtime.Internal.Connection {
         internal async Task SendCommand(GenericCommand command) {
             LCLogger.Debug($"{id} => {FormatCommand(command)}");
             byte[] bytes = command.ToByteArray();
-            Task sendTask = client.Send(bytes);
+            Task sendTask = ws.Send(bytes);
             if (await Task.WhenAny(sendTask, Task.Delay(SEND_TIMEOUT)) == sendTask) {
                 await sendTask;
             } else {
@@ -143,18 +155,22 @@ namespace LeanCloud.Realtime.Internal.Connection {
         }
 
         /// <summary>
-        /// 关闭连接
+        /// 断开连接
         /// </summary>
-        /// <returns></returns>
+        private void Disconnect() {
-        internal async Task Close() {
+            state = State.Closed;
-            OnNotification = null;
-            OnDisconnect = null;
-            OnReconnected = null;
             heartBeat.Stop();
-            await client.Close();
+            _ = ws.Close();
+            foreach (LCIMClient client in idToClients.Values) {
+                client.HandleDisconnected();
+            }
         }
 
-        private void OnClientMessage(byte[] bytes) {
+        /// <summary>
+        /// 消息接收回调
+        /// </summary>
+        /// <param name="bytes"></param>
+        private void OnMessage(byte[] bytes) {
             try {
                 GenericCommand command = GenericCommand.Parser.ParseFrom(bytes);
                 LCLogger.Debug($"{id} <= {FormatCommand(command)}");
@@ -179,10 +195,17 @@ namespace LeanCloud.Realtime.Internal.Connection {
                     }
                 } else {
                     if (command.Cmd == CommandType.Echo) {
+                        // 心跳应答
                         heartBeat.Pong();
+                    } else if (command.Cmd == CommandType.Goaway) {
+                        // 针对连接的消息
+                        Reset();
                     } else {
                         // 通知
-                        OnNotification?.Invoke(command);
+                        if (idToClients.TryGetValue(command.PeerId, out LCIMClient client)) {
+                            // 通知具体客户端
+                            client.HandleNotification(command);
+                        }
                     }
                 }
             } catch (Exception e) {
@@ -190,17 +213,27 @@ namespace LeanCloud.Realtime.Internal.Connection {
             }
         }
 
-        private void OnClientDisconnect() {
+        /// <summary>
-            heartBeat.Stop();
+        /// 连接断开回调
-            OnDisconnect?.Invoke();
+        /// </summary>
-            // 重连
+        private void OnDisconnect() {
+            Disconnect();
             _ = Reconnect();
         }
 
-        private async void OnPingTimeout() {
+        /// <summary>
-            await client.Close();
+        /// 重置连接
-            OnDisconnect?.Invoke();
+        /// </summary>
-            // 重连
+        /// <returns></returns>
+        internal void Reset() {
+            Disconnect();
+            // 重新创建连接组件
+            heartBeat = new LCHeartBeat(this, OnDisconnect);
+            router = new LCRTMRouter();
+            ws = new LCWebSocketClient {
+                OnMessage = OnMessage,
+                OnClose = OnDisconnect
+            };
             _ = Reconnect();
         }
 
@@ -223,9 +256,11 @@ namespace LeanCloud.Realtime.Internal.Connection {
                 if (reconnectCount < MAX_RECONNECT_TIMES) {
                     // 重连成功
                     LCLogger.Debug("Reconnected");
-                    client.OnMessage = OnClientMessage;
+                    ws.OnMessage = OnMessage;
-                    client.OnClose = OnClientDisconnect;
+                    ws.OnClose = OnDisconnect;
-                    OnReconnected?.Invoke();
+                    foreach (LCIMClient client in idToClients.Values) {
+                        client.HandleReconnected();
+                    }
                     break;
                 } else {
                     // 重置 Router，继续尝试重连
@@ -242,5 +277,31 @@ namespace LeanCloud.Realtime.Internal.Connection {
             sb.Append($"\n{command}");
             return sb.ToString();
         }
+
+        internal void Register(LCIMClient client) {
+            idToClients[client.Id] = client;
+        }
+
+        internal void UnRegister(LCIMClient client) {
+            idToClients.Remove(client.Id);
+            if (idToClients.Count == 0) {
+                Disconnect();
+                LCRealtime.RemoveConnection(this);
+            }
+        }
+
+        /// <summary>
+        /// 暂停连接
+        /// </summary>
+        internal void Pause() {
+            Disconnect();
+        }
+
+        /// <summary>
+        ///  恢复连接
+        /// </summary>
+        internal void Resume() {
+            _ = Reconnect();
+        }
     }
 }

Realtime/Realtime/Internal/Connection/LCHeartBeat.cs

@@ -11,7 +11,7 @@ namespace LeanCloud.Realtime.Internal.Connection {
     /// 3. 每隔 180s 检测 pong 包间隔，超过 360s 则认为断开
     /// </summary>
     public class LCHeartBeat {
-        private const int PING_INTERVAL = 5 * 1000;
+        private const int PING_INTERVAL = 180 * 1000;
 
         private readonly LCConnection connection;
 

Realtime/Realtime/Internal/Controller/LCIMController.cs

@@ -1,5 +1,4 @@
-using System.Threading.Tasks;
+using LeanCloud.Realtime.Internal.Protocol;
-using LeanCloud.Realtime.Internal.Protocol;
 using LeanCloud.Realtime.Internal.Connection;
 
 namespace LeanCloud.Realtime.Internal.Controller {
@@ -12,11 +11,11 @@ namespace LeanCloud.Realtime.Internal.Controller {
             Client = client;
         }
 
-        internal abstract Task OnNotification(GenericCommand notification);
+        internal abstract void HandleNotification(GenericCommand notification);
 
         protected LCConnection Connection {
             get {
-                return Client.Connection;
+                return LCRealtime.GetConnection(LCApplication.AppId);
             }
         }
 

Realtime/Realtime/Internal/Controller/LCIMConversationController.cs

@@ -125,7 +125,7 @@ namespace LeanCloud.Realtime.Internal.Controller {
             };
             GenericCommand request = NewCommand(CommandType.Conv, OpType.Update);
             request.ConvMessage = conv;
-            GenericCommand response = await Client.Connection.SendRequest(request);
+            GenericCommand response = await Connection.SendRequest(request);
             JsonObjectMessage attr = response.ConvMessage.AttrModified;
             // 更新自定义属性
             if (attr != null) {
@@ -159,7 +159,7 @@ namespace LeanCloud.Realtime.Internal.Controller {
             }
             GenericCommand request = NewCommand(CommandType.Conv, OpType.Add);
             request.ConvMessage = conv;
-            GenericCommand response = await Client.Connection.SendRequest(request);
+            GenericCommand response = await Connection.SendRequest(request);
             List<string> allowedIds = response.ConvMessage.AllowedPids.ToList();
             List<ErrorCommand> errors = response.ConvMessage.FailedPids.ToList();
             return NewPartiallySuccessResult(allowedIds, errors);
@@ -189,7 +189,7 @@ namespace LeanCloud.Realtime.Internal.Controller {
             }
             GenericCommand request = NewCommand(CommandType.Conv, OpType.Remove);
             request.ConvMessage = conv;
-            GenericCommand response = await Client.Connection.SendRequest(request);
+            GenericCommand response = await Connection.SendRequest(request);
             List<string> allowedIds = response.ConvMessage.AllowedPids.ToList();
             List<ErrorCommand> errors = response.ConvMessage.FailedPids.ToList();
             return NewPartiallySuccessResult(allowedIds, errors);
@@ -206,7 +206,7 @@ namespace LeanCloud.Realtime.Internal.Controller {
             };
             GenericCommand request = NewCommand(CommandType.Conv, OpType.Mute);
             request.ConvMessage = conv;
-            await Client.Connection.SendRequest(request);
+            await Connection.SendRequest(request);
         }
 
         /// <summary>
@@ -220,7 +220,7 @@ namespace LeanCloud.Realtime.Internal.Controller {
             };
             GenericCommand request = NewCommand(CommandType.Conv, OpType.Unmute);
             request.ConvMessage = conv;
-            await Client.Connection.SendRequest(request);
+            await Connection.SendRequest(request);
         }
 
         /// <summary>
@@ -240,7 +240,7 @@ namespace LeanCloud.Realtime.Internal.Controller {
             conv.M.AddRange(clientIds);
             GenericCommand request = NewCommand(CommandType.Conv, OpType.AddShutup);
             request.ConvMessage = conv;
-            GenericCommand response = await Client.Connection.SendRequest(request);
+            GenericCommand response = await Connection.SendRequest(request);
             return NewPartiallySuccessResult(response.ConvMessage.AllowedPids, response.ConvMessage.FailedPids);
         }
 
@@ -258,7 +258,7 @@ namespace LeanCloud.Realtime.Internal.Controller {
             conv.M.AddRange(clientIds);
             GenericCommand request = NewCommand(CommandType.Conv, OpType.RemoveShutup);
             request.ConvMessage = conv;
-            GenericCommand response = await Client.Connection.SendRequest(request);
+            GenericCommand response = await Connection.SendRequest(request);
             return NewPartiallySuccessResult(response.ConvMessage.AllowedPids, response.ConvMessage.FailedPids);
         }
 
@@ -285,7 +285,7 @@ namespace LeanCloud.Realtime.Internal.Controller {
             }
             GenericCommand request = NewCommand(CommandType.Blacklist, OpType.Block);
             request.BlacklistMessage = blacklist;
-            GenericCommand response = await Client.Connection.SendRequest(request);
+            GenericCommand response = await Connection.SendRequest(request);
             return NewPartiallySuccessResult(response.BlacklistMessage.AllowedPids, response.BlacklistMessage.FailedPids);
         }
 
@@ -312,7 +312,7 @@ namespace LeanCloud.Realtime.Internal.Controller {
             }
             GenericCommand request = NewCommand(CommandType.Blacklist, OpType.Unblock);
             request.BlacklistMessage = blacklist;
-            GenericCommand response = await Client.Connection.SendRequest(request);
+            GenericCommand response = await Connection.SendRequest(request);
             return NewPartiallySuccessResult(response.BlacklistMessage.AllowedPids, response.BlacklistMessage.FailedPids);
         }
 
@@ -336,7 +336,7 @@ namespace LeanCloud.Realtime.Internal.Controller {
             };
             GenericCommand request = NewCommand(CommandType.Conv, OpType.MemberInfoUpdate);
             request.ConvMessage = conv;
-            GenericCommand response = await Client.Connection.SendRequest(request);
+            GenericCommand response = await Connection.SendRequest(request);
         }
 
         /// <summary>
@@ -386,7 +386,7 @@ namespace LeanCloud.Realtime.Internal.Controller {
             }
             GenericCommand request = NewCommand(CommandType.Conv, OpType.QueryShutup);
             request.ConvMessage = conv;
-            GenericCommand response = await Client.Connection.SendRequest(request);
+            GenericCommand response = await Connection.SendRequest(request);
             return new LCIMPageResult {
                 Results = new ReadOnlyCollection<string>(response.ConvMessage.M),
                 Next = response.ConvMessage.Next
@@ -412,7 +412,7 @@ namespace LeanCloud.Realtime.Internal.Controller {
             }
             GenericCommand request = NewCommand(CommandType.Blacklist, OpType.Query);
             request.BlacklistMessage = black;
-            GenericCommand response = await Client.Connection.SendRequest(request);
+            GenericCommand response = await Connection.SendRequest(request);
             return new LCIMPageResult {
                 Results = new ReadOnlyCollection<string>(response.BlacklistMessage.BlockedPids),
                 Next = response.BlacklistMessage.Next
@@ -525,7 +525,7 @@ namespace LeanCloud.Realtime.Internal.Controller {
             };
             GenericCommand request = NewCommand(CommandType.Conv, OpType.Members);
             request.ConvMessage = conv;
-            GenericCommand response = await Client.Connection.SendRequest(request);
+            GenericCommand response = await Connection.SendRequest(request);
             ReadOnlyCollection<string> members = response.ConvMessage.M
                 .ToList().AsReadOnly();
             return members;
@@ -541,7 +541,7 @@ namespace LeanCloud.Realtime.Internal.Controller {
             conv.Cids.Add(convId);
             GenericCommand request = NewCommand(CommandType.Conv, OpType.IsMember);
             request.ConvMessage = conv;
-            GenericCommand response = await Client.Connection.SendRequest(request);
+            GenericCommand response = await Connection.SendRequest(request);
             JsonObjectMessage jsonObj = response.ConvMessage.Results;
             Dictionary<string, object> result = JsonConvert.DeserializeObject<Dictionary<string, object>>(jsonObj.Data);
             if (result.TryGetValue(convId, out object obj)) {
@@ -573,11 +573,11 @@ namespace LeanCloud.Realtime.Internal.Controller {
 
         #region 消息处理
 
-        internal override async Task OnNotification(GenericCommand notification) {
+        internal override void HandleNotification(GenericCommand notification) {
             if (notification.Cmd == CommandType.Conv) {
-                await OnConversation(notification);
+                _ = OnConversation(notification);
             } else if (notification.Cmd == CommandType.Unread) {
-                await OnUnread(notification);
+                _ = OnUnread(notification);
             }
         }
 
@@ -695,8 +695,8 @@ namespace LeanCloud.Realtime.Internal.Controller {
         /// <returns></returns>
         private async Task OnLeft(ConvCommand convMessage) {
             LCIMConversation conversation = await Client.GetOrQueryConversation(convMessage.Cid);
-            // TODO 从内存中清除对话
+            // 从内存中清除对话
-
+            Client.ConversationDict.Remove(conversation.Id);
             Client.OnKicked?.Invoke(conversation, convMessage.InitBy);
         }
 

Realtime/Realtime/Internal/Controller/LCIMGoAwayController.cs

@@ -1,19 +0,0 @@
-using System.Threading.Tasks;
-using LeanCloud.Realtime.Internal.Protocol;
-
-namespace LeanCloud.Realtime.Internal.Controller {
-    internal class LCIMGoAwayController : LCIMController {
-        internal LCIMGoAwayController(LCIMClient client) : base(client) {
-
-        }
-
-        #region 消息处理
-
-        internal override async Task OnNotification(GenericCommand notification) {
-            // 清空缓存，断开连接，等待重新连接
-            await Connection.Reset();
-        }
-
-        #endregion
-    }
-}

Realtime/Realtime/Internal/Controller/LCIMMessageController.cs

@@ -65,7 +65,7 @@ namespace LeanCloud.Realtime.Internal.Controller {
             if (command.Priority > 0) {
                 command.Priority = (int)options.Priority;
             }
-            GenericCommand response = await Client.Connection.SendRequest(command);
+            GenericCommand response = await Connection.SendRequest(command);
             // 消息发送应答
             AckCommand ack = response.AckMessage;
             message.Id = ack.Uid;
@@ -94,7 +94,7 @@ namespace LeanCloud.Realtime.Internal.Controller {
             patch.Patches.Add(item);
             GenericCommand request = NewCommand(CommandType.Patch, OpType.Modify);
             request.PatchMessage = patch;
-            await Client.Connection.SendRequest(request);
+            await Connection.SendRequest(request);
         }
 
         /// <summary>
@@ -130,7 +130,7 @@ namespace LeanCloud.Realtime.Internal.Controller {
             patch.Patches.Add(item);
             GenericCommand request = NewCommand(CommandType.Patch, OpType.Modify);
             request.PatchMessage = patch;
-            GenericCommand response = await Client.Connection.SendRequest(request);
+            GenericCommand response = await Connection.SendRequest(request);
         }
 
         /// <summary>
@@ -170,7 +170,7 @@ namespace LeanCloud.Realtime.Internal.Controller {
             }
             GenericCommand request = NewCommand(CommandType.Logs, OpType.Open);
             request.LogsMessage = logs;
-            GenericCommand response = await Client.Connection.SendRequest(request);
+            GenericCommand response = await Connection.SendRequest(request);
             // 反序列化聊天记录
             return response.LogsMessage.Logs.Select(item => {
                 LCIMMessage message;
@@ -211,7 +211,7 @@ namespace LeanCloud.Realtime.Internal.Controller {
             };
             GenericCommand command = NewCommand(CommandType.Ack);
             command.AckMessage = ack;
-            await Client.Connection.SendCommand(command);
+            await Connection.SendCommand(command);
         }
 
         /// <summary>
@@ -231,20 +231,20 @@ namespace LeanCloud.Realtime.Internal.Controller {
             read.Convs.Add(tuple);
             GenericCommand command = NewCommand(CommandType.Read);
             command.ReadMessage = read;
-            await Client.Connection.SendCommand(command);
+            await Connection.SendCommand(command);
         }
 
         #endregion
 
         #region 消息处理
 
-        internal override async Task OnNotification(GenericCommand notification) {
+        internal override void HandleNotification(GenericCommand notification) {
             if (notification.Cmd == CommandType.Direct) {
-                await OnMessaage(notification);
+                _ = OnMessaage(notification);
             } else if (notification.Cmd == CommandType.Patch) {
-                await OnMessagePatched(notification);
+                _ = OnMessagePatched(notification);
             } else if (notification.Cmd == CommandType.Rcp) {
-                await OnMessageReceipt(notification);
+                _ = OnMessageReceipt(notification);
             }
         }
 

Realtime/Realtime/Internal/Controller/LCIMSessionController.cs

@@ -20,13 +20,16 @@ namespace LeanCloud.Realtime.Internal.Controller {
         /// </summary>
         /// <returns></returns>
         internal async Task Open(bool force) {
+            await Connection.Connect();
+
             SessionCommand session = await NewSessionCommand();
             session.R = !force;
             session.ConfigBitmap = 0x2B;
             GenericCommand request = NewCommand(CommandType.Session, OpType.Open);
             request.SessionMessage = session;
-            GenericCommand response = await Client.Connection.SendRequest(request);
+            GenericCommand response = await Connection.SendRequest(request);
             UpdateSession(response.SessionMessage);
+            Connection.Register(Client);
         }
 
         /// <summary>
@@ -38,11 +41,11 @@ namespace LeanCloud.Realtime.Internal.Controller {
             session.R = true;
             GenericCommand request = NewCommand(CommandType.Session, OpType.Open);
             request.SessionMessage = session;
-            GenericCommand response = await Client.Connection.SendRequest(request);
+            GenericCommand response = await Connection.SendRequest(request);
             if (response.Op == OpType.Opened) {
                 UpdateSession(response.SessionMessage);
             } else if (response.Op == OpType.Closed) {
-                await OnClosed(response.SessionMessage);
+                OnClosed(response.SessionMessage);
             }
         }
 
@@ -52,7 +55,8 @@ namespace LeanCloud.Realtime.Internal.Controller {
         /// <returns></returns>
         internal async Task Close() {
             GenericCommand request = NewCommand(CommandType.Session, OpType.Close);
-            await Client.Connection.SendRequest(request);
+            await Connection.SendRequest(request);
+            Connection.UnRegister(Client);
         }
 
         /// <summary>
@@ -72,7 +76,7 @@ namespace LeanCloud.Realtime.Internal.Controller {
             SessionCommand session = await NewSessionCommand();
             GenericCommand request = NewCommand(CommandType.Session, OpType.Refresh);
             request.SessionMessage = session;
-            GenericCommand response = await Client.Connection.SendRequest(request);
+            GenericCommand response = await Connection.SendRequest(request);
             UpdateSession(response.SessionMessage);
         }
 
@@ -120,10 +124,10 @@ namespace LeanCloud.Realtime.Internal.Controller {
 
         #region 消息处理
 
-        internal override async Task OnNotification(GenericCommand notification) {
+        internal override void HandleNotification(GenericCommand notification) {
             switch (notification.Op) {
                 case OpType.Closed:
-                    await OnClosed(notification.SessionMessage);
+                    OnClosed(notification.SessionMessage);
                     break;
                 default:
                     break;
@@ -135,11 +139,11 @@ namespace LeanCloud.Realtime.Internal.Controller {
         /// </summary>
         /// <param name="session"></param>
         /// <returns></returns>
-        private async Task OnClosed(SessionCommand session) {
+        private void OnClosed(SessionCommand session) {
             int code = session.Code;
             string reason = session.Reason;
             string detail = session.Detail;
-            await Connection.Close();
+            Connection.UnRegister(Client);
             Client.OnClose?.Invoke(code, reason);
         }
 

Realtime/Realtime/Internal/WebSocket/LCWebSocketClient.cs

@@ -81,7 +81,7 @@ namespace LeanCloud.Realtime.Internal.WebSocket {
         }
 
         /// <summary>
-        /// 发送数据
+        /// 发送二进制数据
         /// </summary>
         /// <param name="data"></param>
         /// <returns></returns>

Realtime/Realtime/LCIMClient.cs

@@ -3,31 +3,44 @@ using System.Collections.Generic;
 using System.Threading.Tasks;
 using System.Linq;
 using System.Collections.ObjectModel;
-using LeanCloud.Common;
 using LeanCloud.Storage;
 using LeanCloud.Realtime.Internal.Protocol;
 using LeanCloud.Realtime.Internal.Controller;
-using LeanCloud.Realtime.Internal.Connection;
 
 namespace LeanCloud.Realtime {
     /// <summary>
     /// 通信客户端
     /// </summary>
     public class LCIMClient {
+        /// <summary>
+        /// 对话缓存
+        /// </summary>
         internal Dictionary<string, LCIMConversation> ConversationDict;
 
+        /// <summary>
+        /// 用户 Id
+        /// </summary>
         public string Id {
             get; private set;
         }
 
+        /// <summary>
+        /// 用户标识
+        /// </summary>
         public string Tag {
             get; private set;
         }
 
+        /// <summary>
+        /// 设备 Id
+        /// </summary>
         public string DeviceId {
             get; private set;
         }
 
+        /// <summary>
+        /// 登录 tokens
+        /// </summary>
         internal string SessionToken {
             get; private set;
         }
@@ -219,10 +232,6 @@ namespace LeanCloud.Realtime {
             get; private set;
         }
 
-        internal LCConnection Connection {
-            get; set;
-        }
-
         internal LCIMSessionController SessionController {
             get; private set;
         }
@@ -231,10 +240,6 @@ namespace LeanCloud.Realtime {
             get; private set;
         }
 
-        internal LCIMGoAwayController GoAwayController {
-            get; private set;
-        }
-
         internal LCIMConversationController ConversationController {
             get; private set;
         }
@@ -281,13 +286,6 @@ namespace LeanCloud.Realtime {
             SessionController = new LCIMSessionController(this);
             ConversationController = new LCIMConversationController(this);
             MessageController = new LCIMMessageController(this);
-            GoAwayController = new LCIMGoAwayController(this);
-
-            Connection = new LCConnection(Id) {
-                OnNotification = OnConnectionNotification,
-                OnDisconnect = OnConnectionDisconnect,
-                OnReconnected = OnConnectionReconnect
-            };
         }
 
         /// <summary>
@@ -296,14 +294,12 @@ namespace LeanCloud.Realtime {
         /// <param name="force">是否强制登录</param>
         /// <returns></returns>
         public async Task Open(bool force = true) {
-            await Connection.Connect();
             try {
                 // 打开 Session
                 await SessionController.Open(force);
             } catch (Exception e) {
                 LCLogger.Error(e);
                 // 如果 session 阶段异常，则关闭连接
-                await Connection.Close();
                 throw e;
             }
         }
@@ -315,7 +311,6 @@ namespace LeanCloud.Realtime {
         public async Task Close() {
             // 关闭 session
             await SessionController.Close();
-            await Connection.Close();
         }
 
         /// <summary>
@@ -437,37 +432,30 @@ namespace LeanCloud.Realtime {
 
         #endregion
 
-        private void OnConnectionNotification(GenericCommand notification) {
+        internal void HandleNotification(GenericCommand notification) {
             switch (notification.Cmd) {
                 case CommandType.Session:
-                    _ = SessionController.OnNotification(notification);
+                    SessionController.HandleNotification(notification);
                     break;
                 case CommandType.Conv:
                 case CommandType.Unread:
-                    _ = ConversationController.OnNotification(notification);
+                    ConversationController.HandleNotification(notification);
                     break;
                 case CommandType.Direct:
                 case CommandType.Patch:
                 case CommandType.Rcp:
-                    _ = MessageController.OnNotification(notification);
+                    MessageController.HandleNotification(notification);
-                    break;
-                case CommandType.Goaway:
-                    _ = GoAwayController.OnNotification(notification);
                     break;
                 default:
                     break;
             }
         }
 
-        private void OnConnectionDisconnect() {
+        internal void HandleDisconnected() {
             OnPaused?.Invoke();
         }
 
-        private void OnConnectionReconnect() {
+        internal async void HandleReconnected() {
-            _ = HandleReconnected();
-        }
-
-        private async Task HandleReconnected() {
             try {
                 // 打开 Session
                 await SessionController.Reopen();
@@ -475,8 +463,7 @@ namespace LeanCloud.Realtime {
                 OnResume?.Invoke();
             } catch (Exception e) {
                 LCLogger.Error(e);
-                await Connection.Close();
+                // 重连成功，但 session/open 失败
-                // TODO 告知
                 OnClose?.Invoke(0, string.Empty);
             }
         }

Realtime/Realtime/LCRealtime.cs

@@ -0,0 +1,51 @@
+using System.Collections.Generic;
+using LeanCloud.Realtime.Internal.Connection;
+
+namespace LeanCloud.Realtime {
+    public class LCRealtime {
+        /// <summary>
+        /// RTM 服务中，每个 app 对应一条连接
+        /// </summary>
+        private static readonly Dictionary<string, LCConnection> appToConnections = new Dictionary<string, LCConnection>();
+
+        /// <summary>
+        /// 获取对应的 Connection
+        /// </summary>
+        /// <param name="appId"></param>
+        /// <returns></returns>
+        internal static LCConnection GetConnection(string appId) {
+            if (appToConnections.TryGetValue(appId, out LCConnection connection)) {
+                return connection;
+            }
+            connection = new LCConnection(appId);
+            appToConnections[appId] = connection;
+            return connection;
+        }
+
+        /// <summary>
+        /// 移除 Connection
+        /// </summary>
+        /// <param name="connection"></param>
+        internal static void RemoveConnection(LCConnection connection) {
+            appToConnections.Remove(connection.id);
+        }
+
+        /// <summary>
+        /// 主动断开所有 RTM 连接
+        /// </summary>
+        public static void Pause() {
+            foreach (LCConnection connection in appToConnections.Values) {
+                connection.Pause();
+            }
+        }
+
+        /// <summary>
+        /// 主动恢复所有 RTM 连接
+        /// </summary>
+        public static void Resume() {
+            foreach (LCConnection connection in appToConnections.Values) {
+                connection.Resume();
+            }
+        }
+    }
+}

Sample/RealtimeApp/Program.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Threading.Tasks;
 using LeanCloud;
 using LeanCloud.Realtime;
 
@@ -7,15 +8,43 @@ using static System.Console;
 namespace RealtimeApp {
     class Program {
         static void Main(string[] args) {
-            Console.WriteLine("Hello World!");
+            WriteLine("Hello World!");
 
             SingleThreadSynchronizationContext.Run(async () => {
                 LCLogger.LogDelegate += Print;
                 LCApplication.Initialize("ikGGdRE2YcVOemAaRbgp1xGJ-gzGzoHsz", "NUKmuRbdAhg1vrb2wexYo1jo", "https://ikggdre2.lc-cn-n1-shared.com");
 
-                LCIMClient client = new LCIMClient("lean");
+                LCIMClient client = new LCIMClient("lean") {
+                    OnPaused = () => {
+                        WriteLine("~~~~~~~~~~~~~~~ disconnected");
+                    },
+                    OnResume = () => {
+                        WriteLine("~~~~~~~~~~~~~~~ reconnected");
+                    }
+                };
+
                 await client.Open();
-                //await client.Close();
+
+                int count = 0;
+                while (count < 2) {
+                    WriteLine($"pause : {count}");
+
+                    await Task.Delay(5 * 1000);
+                    LCRealtime.Pause();
+
+                    await Task.Delay(5 * 1000);
+                    LCRealtime.Resume();
+
+                    await Task.Delay(5 * 1000);
+                    count++;
+                }
+
+                try {
+                    await client.Close();
+                    // Done
+                } catch (Exception e) {
+                    WriteLine($"xxxxxxxxxxxx {e.Message}");
+                }
             });
         }