From 5891f9287b925279285fd67fe7c1beef4de982bd Mon Sep 17 00:00:00 2001 From: oneRain Date: Mon, 22 Jun 2020 18:13:26 +0800 Subject: [PATCH 1/9] =?UTF-8?q?chore:=20=E7=9B=B8=E5=90=8C=20app=20?= =?UTF-8?q?=E5=85=B1=E4=BA=AB=E8=BF=9E=E6=8E=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../Internal/Connection/LCConnection.cs | 47 +++++++++++++++++-- Realtime/Realtime/LCIMClient.cs | 12 ++--- Realtime/Realtime/LCRealtime.cs | 40 ++++++++++++++++ 3 files changed, 89 insertions(+), 10 deletions(-) create mode 100644 Realtime/Realtime/LCRealtime.cs diff --git a/Realtime/Realtime/Internal/Connection/LCConnection.cs b/Realtime/Realtime/Internal/Connection/LCConnection.cs index 565d266..4a260d5 100644 --- a/Realtime/Realtime/Internal/Connection/LCConnection.cs +++ b/Realtime/Realtime/Internal/Connection/LCConnection.cs @@ -12,6 +12,28 @@ namespace LeanCloud.Realtime.Internal.Connection { /// 连接层，只与数据协议相关 /// public class LCConnection { + /// + /// 连接状态 + /// + enum State { + /// + /// 初始状态 + /// + None, + /// + /// 连接中 + /// + Connecting, + /// + /// 连接成功 + /// + Open, + /// + /// 关闭的 + /// + Closed, + } + /// /// 发送超时 /// @@ -62,6 +84,9 @@ namespace LeanCloud.Realtime.Internal.Connection { private LCWebSocketClient client; + private State state; + private Task connectTask; + internal LCConnection(string id) { this.id = id; responses = new Dictionary>(); @@ -71,9 +96,22 @@ namespace LeanCloud.Realtime.Internal.Connection { OnMessage = OnClientMessage, OnClose = OnClientDisconnect }; + state = State.None; } - internal async Task Connect() { + internal Task Connect() { + if (state == State.Open) { + return Task.FromResult(null); + } + if (state == State.Connecting) { + return connectTask; + } + connectTask = _Connect(); + return connectTask; + } + + internal async Task _Connect() { + state = State.Connecting; try { LCRTMServer rtmServer = await router.GetServer(); try { @@ -86,6 +124,7 @@ namespace LeanCloud.Realtime.Internal.Connection { } // 启动心跳 heartBeat.Start(); + state = State.Open; } catch (Exception e) { throw e; } @@ -191,14 +230,16 @@ namespace LeanCloud.Realtime.Internal.Connection { } private void OnClientDisconnect() { + state = State.Closed; heartBeat.Stop(); OnDisconnect?.Invoke(); // 重连 _ = Reconnect(); } - private async void OnPingTimeout() { - await client.Close(); + private void OnPingTimeout() { + state = State.Closed; + _ = client.Close(); OnDisconnect?.Invoke(); // 重连 _ = Reconnect(); diff --git a/Realtime/Realtime/LCIMClient.cs b/Realtime/Realtime/LCIMClient.cs index 4799c4a..dbb5463 100644 --- a/Realtime/Realtime/LCIMClient.cs +++ b/Realtime/Realtime/LCIMClient.cs @@ -3,7 +3,6 @@ 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; @@ -283,11 +282,10 @@ namespace LeanCloud.Realtime { MessageController = new LCIMMessageController(this); GoAwayController = new LCIMGoAwayController(this); - Connection = new LCConnection(Id) { - OnNotification = OnConnectionNotification, - OnDisconnect = OnConnectionDisconnect, - OnReconnected = OnConnectionReconnect - }; + Connection = LCRealtime.GetConnection(LCApplication.AppId); + Connection.OnNotification = OnConnectionNotification; + Connection.OnDisconnect = OnConnectionDisconnect; + Connection.OnReconnected = OnConnectionReconnect; } /// @@ -315,7 +313,7 @@ namespace LeanCloud.Realtime { public async Task Close() { // 关闭 session await SessionController.Close(); - await Connection.Close(); + //await Connection.Close(); } /// diff --git a/Realtime/Realtime/LCRealtime.cs b/Realtime/Realtime/LCRealtime.cs new file mode 100644 index 0000000..bd313b3 --- /dev/null +++ b/Realtime/Realtime/LCRealtime.cs @@ -0,0 +1,40 @@ +using System.Collections.Generic; +using LeanCloud.Realtime.Internal.Connection; + +namespace LeanCloud.Realtime { + public class LCRealtime { + /// + /// RTM 服务中，每个 app 对应一条连接 + /// + private static readonly Dictionary appToConnections = new Dictionary(); + + /// + /// 获取对应的 Connection + /// + /// + /// + internal static LCConnection GetConnection(string appId) { + if (appToConnections.TryGetValue(appId, out LCConnection connection)) { + return connection; + } + string connId = appId.Substring(0, 8).ToLower(); + connection = new LCConnection(connId); + appToConnections[appId] = connection; + return connection; + } + + /// + /// 主动断开所有 RTM 连接 + /// + public static void Pause() { + + } + + /// + /// 主动恢复所有 RTM 连接 + /// + public static void Resume() { + + } + } +} From 2c919d43449d64a86ea1685492a7f05b62de464d Mon Sep 17 00:00:00 2001 From: oneRain Date: Tue, 23 Jun 2020 15:25:30 +0800 Subject: [PATCH 2/9] =?UTF-8?q?chore:=20=E6=94=AF=E6=8C=81=E7=9B=B8?= =?UTF-8?q?=E5=90=8C=20app=20=E5=85=B1=E4=BA=AB=E8=BF=9E=E6=8E=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- Realtime/Realtime-Unity/Realtime-Unity.csproj | 3 + Realtime/Realtime.Test/Client.cs | 31 +++++---- Realtime/Realtime.Test/Conversation.cs | 4 +- Realtime/Realtime.Test/Utils.cs | 6 +- .../Internal/Connection/LCConnection.cs | 68 ++++++++++++++----- .../Internal/Connection/LCHeartBeat.cs | 2 +- .../Internal/Controller/LCIMController.cs | 5 +- .../Controller/LCIMConversationController.cs | 6 +- .../Controller/LCIMGoAwayController.cs | 4 +- .../Controller/LCIMMessageController.cs | 8 +-- .../Controller/LCIMSessionController.cs | 10 +-- .../Internal/WebSocket/LCWebSocketClient.cs | 2 +- Realtime/Realtime/LCIMClient.cs | 48 ++++++++----- Realtime/Realtime/LCRealtime.cs | 11 ++- 14 files changed, 134 insertions(+), 74 deletions(-) diff --git a/Realtime/Realtime-Unity/Realtime-Unity.csproj b/Realtime/Realtime-Unity/Realtime-Unity.csproj index 19ba882..b900418 100644 --- a/Realtime/Realtime-Unity/Realtime-Unity.csproj +++ b/Realtime/Realtime-Unity/Realtime-Unity.csproj @@ -115,6 +115,9 @@ LCIMClient.cs + + LCRealtime.cs + diff --git a/Realtime/Realtime.Test/Client.cs b/Realtime/Realtime.Test/Client.cs index 9669bea..e43a7bf 100644 --- a/Realtime/Realtime.Test/Client.cs +++ b/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"); - await client.Open(); - await client.Close(); + LCIMClient c1 = new LCIMClient("c1"); + LCIMClient c2 = new LCIMClient("c2"); + 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 tcs = new TaskCompletionSource(); - 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 onlineMembers = await chatRoom.GetOnlineMembers(); @@ -105,8 +110,6 @@ namespace Realtime.Test { await client.Close(); await visitor.Close(); - - await tcs.Task; } [Test] diff --git a/Realtime/Realtime.Test/Conversation.cs b/Realtime/Realtime.Test/Conversation.cs index 26ebeb6..31f34cc 100644 --- a/Realtime/Realtime.Test/Conversation.cs +++ b/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; } } } diff --git a/Realtime/Realtime.Test/Utils.cs b/Realtime/Realtime.Test/Utils.cs index 10a85a0..70aa219 100644 --- a/Realtime/Realtime.Test/Utils.cs +++ b/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); diff --git a/Realtime/Realtime/Internal/Connection/LCConnection.cs b/Realtime/Realtime/Internal/Connection/LCConnection.cs index 4a260d5..2e2cb8f 100644 --- a/Realtime/Realtime/Internal/Connection/LCConnection.cs +++ b/Realtime/Realtime/Internal/Connection/LCConnection.cs @@ -82,20 +82,23 @@ namespace LeanCloud.Realtime.Internal.Connection { private LCHeartBeat heartBeat; - private LCWebSocketClient client; + private LCWebSocketClient ws; private State state; private Task connectTask; + private readonly Dictionary clients; + internal LCConnection(string id) { this.id = id; responses = new Dictionary>(); heartBeat = new LCHeartBeat(this, OnPingTimeout); router = new LCRTMRouter(); - client = new LCWebSocketClient { + ws = new LCWebSocketClient { OnMessage = OnClientMessage, OnClose = OnClientDisconnect }; + clients = new Dictionary(); state = State.None; } @@ -106,21 +109,21 @@ namespace LeanCloud.Realtime.Internal.Connection { if (state == State.Connecting) { return connectTask; } - connectTask = _Connect(); + connectTask = ConnectInternal(); return connectTask; } - internal async Task _Connect() { + 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(); @@ -137,11 +140,11 @@ namespace LeanCloud.Realtime.Internal.Connection { internal async Task Reset() { heartBeat?.Stop(); // 关闭就连接 - await client.Close(); + await ws.Close(); // 重新创建连接组件 heartBeat = new LCHeartBeat(this, OnPingTimeout); router = new LCRTMRouter(); - client = new LCWebSocketClient { + ws = new LCWebSocketClient { OnMessage = OnClientMessage, OnClose = OnClientDisconnect }; @@ -173,7 +176,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 { @@ -186,11 +189,12 @@ namespace LeanCloud.Realtime.Internal.Connection { /// /// internal async Task Close() { + LCRealtime.RemoveConnection(this); OnNotification = null; OnDisconnect = null; OnReconnected = null; heartBeat.Stop(); - await client.Close(); + await ws.Close(); } private void OnClientMessage(byte[] bytes) { @@ -221,7 +225,10 @@ namespace LeanCloud.Realtime.Internal.Connection { heartBeat.Pong(); } else { // 通知 - OnNotification?.Invoke(command); + if (clients.TryGetValue(command.PeerId, out LCIMClient client)) { + // 通知具体客户端 + client.HandleNotification(command); + } } } } catch (Exception e) { @@ -232,15 +239,19 @@ namespace LeanCloud.Realtime.Internal.Connection { private void OnClientDisconnect() { state = State.Closed; heartBeat.Stop(); - OnDisconnect?.Invoke(); + foreach (LCIMClient client in clients.Values) { + client.HandleDisconnected(); + } // 重连 _ = Reconnect(); } private void OnPingTimeout() { state = State.Closed; - _ = client.Close(); - OnDisconnect?.Invoke(); + _ = ws.Close(); + foreach (LCIMClient client in clients.Values) { + client.HandleDisconnected(); + } // 重连 _ = Reconnect(); } @@ -264,8 +275,8 @@ namespace LeanCloud.Realtime.Internal.Connection { if (reconnectCount < MAX_RECONNECT_TIMES) { // 重连成功 LCLogger.Debug("Reconnected"); - client.OnMessage = OnClientMessage; - client.OnClose = OnClientDisconnect; + ws.OnMessage = OnClientMessage; + ws.OnClose = OnClientDisconnect; OnReconnected?.Invoke(); break; } else { @@ -283,5 +294,30 @@ namespace LeanCloud.Realtime.Internal.Connection { sb.Append($"\n{command}"); return sb.ToString(); } + + internal void Register(LCIMClient client) { + clients[client.Id] = client; + } + + internal void UnRegister(LCIMClient client) { + clients.Remove(client.Id); + if (clients.Count == 0) { + _ = Close(); + } + } + + /// + /// 暂停连接 + /// + internal void Pause() { + + } + + /// + /// 恢复连接 + /// + internal void Resume() { + + } } } diff --git a/Realtime/Realtime/Internal/Connection/LCHeartBeat.cs b/Realtime/Realtime/Internal/Connection/LCHeartBeat.cs index 01629f3..bd62a3b 100644 --- a/Realtime/Realtime/Internal/Connection/LCHeartBeat.cs +++ b/Realtime/Realtime/Internal/Connection/LCHeartBeat.cs @@ -11,7 +11,7 @@ namespace LeanCloud.Realtime.Internal.Connection { /// 3. 每隔 180s 检测 pong 包间隔，超过 360s 则认为断开 /// public class LCHeartBeat { - private const int PING_INTERVAL = 5 * 1000; + private const int PING_INTERVAL = 180 * 1000; private readonly LCConnection connection; diff --git a/Realtime/Realtime/Internal/Controller/LCIMController.cs b/Realtime/Realtime/Internal/Controller/LCIMController.cs index 598a526..78f134d 100644 --- a/Realtime/Realtime/Internal/Controller/LCIMController.cs +++ b/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,7 +11,7 @@ namespace LeanCloud.Realtime.Internal.Controller { Client = client; } - internal abstract Task OnNotification(GenericCommand notification); + internal abstract void HandleNotification(GenericCommand notification); protected LCConnection Connection { get { diff --git a/Realtime/Realtime/Internal/Controller/LCIMConversationController.cs b/Realtime/Realtime/Internal/Controller/LCIMConversationController.cs index 68798b0..7a484bd 100644 --- a/Realtime/Realtime/Internal/Controller/LCIMConversationController.cs +++ b/Realtime/Realtime/Internal/Controller/LCIMConversationController.cs @@ -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); } } diff --git a/Realtime/Realtime/Internal/Controller/LCIMGoAwayController.cs b/Realtime/Realtime/Internal/Controller/LCIMGoAwayController.cs index 1bcb57d..35d4518 100644 --- a/Realtime/Realtime/Internal/Controller/LCIMGoAwayController.cs +++ b/Realtime/Realtime/Internal/Controller/LCIMGoAwayController.cs @@ -9,9 +9,9 @@ namespace LeanCloud.Realtime.Internal.Controller { #region 消息处理 - internal override async Task OnNotification(GenericCommand notification) { + internal override void HandleNotification(GenericCommand notification) { // 清空缓存，断开连接，等待重新连接 - await Connection.Reset(); + _ = Connection.Reset(); } #endregion diff --git a/Realtime/Realtime/Internal/Controller/LCIMMessageController.cs b/Realtime/Realtime/Internal/Controller/LCIMMessageController.cs index 0369be1..bfa0099 100644 --- a/Realtime/Realtime/Internal/Controller/LCIMMessageController.cs +++ b/Realtime/Realtime/Internal/Controller/LCIMMessageController.cs @@ -238,13 +238,13 @@ namespace LeanCloud.Realtime.Internal.Controller { #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); } } diff --git a/Realtime/Realtime/Internal/Controller/LCIMSessionController.cs b/Realtime/Realtime/Internal/Controller/LCIMSessionController.cs index 28b913c..73bce54 100644 --- a/Realtime/Realtime/Internal/Controller/LCIMSessionController.cs +++ b/Realtime/Realtime/Internal/Controller/LCIMSessionController.cs @@ -42,7 +42,7 @@ namespace LeanCloud.Realtime.Internal.Controller { if (response.Op == OpType.Opened) { UpdateSession(response.SessionMessage); } else if (response.Op == OpType.Closed) { - await OnClosed(response.SessionMessage); + OnClosed(response.SessionMessage); } } @@ -120,10 +120,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 +135,11 @@ namespace LeanCloud.Realtime.Internal.Controller { /// /// /// - 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); } diff --git a/Realtime/Realtime/Internal/WebSocket/LCWebSocketClient.cs b/Realtime/Realtime/Internal/WebSocket/LCWebSocketClient.cs index ca42dec..ab4277f 100644 --- a/Realtime/Realtime/Internal/WebSocket/LCWebSocketClient.cs +++ b/Realtime/Realtime/Internal/WebSocket/LCWebSocketClient.cs @@ -81,7 +81,7 @@ namespace LeanCloud.Realtime.Internal.WebSocket { } /// - /// 发送数据 + /// 发送二进制数据 /// /// /// diff --git a/Realtime/Realtime/LCIMClient.cs b/Realtime/Realtime/LCIMClient.cs index dbb5463..e4c3acb 100644 --- a/Realtime/Realtime/LCIMClient.cs +++ b/Realtime/Realtime/LCIMClient.cs @@ -13,20 +13,35 @@ namespace LeanCloud.Realtime { /// 通信客户端 /// public class LCIMClient { + /// + /// 对话缓存 + /// internal Dictionary ConversationDict; + /// + /// 用户 Id + /// public string Id { get; private set; } + /// + /// 用户标识 + /// public string Tag { get; private set; } + /// + /// 设备 Id + /// public string DeviceId { get; private set; } + /// + /// 登录 tokens + /// internal string SessionToken { get; private set; } @@ -283,9 +298,6 @@ namespace LeanCloud.Realtime { GoAwayController = new LCIMGoAwayController(this); Connection = LCRealtime.GetConnection(LCApplication.AppId); - Connection.OnNotification = OnConnectionNotification; - Connection.OnDisconnect = OnConnectionDisconnect; - Connection.OnReconnected = OnConnectionReconnect; } /// @@ -298,10 +310,11 @@ namespace LeanCloud.Realtime { try { // 打开 Session await SessionController.Open(force); + Connection.Register(this); } catch (Exception e) { LCLogger.Error(e); // 如果 session 阶段异常，则关闭连接 - await Connection.Close(); + Connection.UnRegister(this); throw e; } } @@ -313,7 +326,7 @@ namespace LeanCloud.Realtime { public async Task Close() { // 关闭 session await SessionController.Close(); - //await Connection.Close(); + Connection.UnRegister(this); } /// @@ -435,37 +448,36 @@ namespace LeanCloud.Realtime { #endregion - private void OnConnectionNotification(GenericCommand notification) { + internal void HandleNotification(GenericCommand notification) { + if (notification.PeerId != Id) { + return; + } 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); + GoAwayController.HandleNotification(notification); break; default: break; } } - private void OnConnectionDisconnect() { + internal void HandleDisconnected() { OnPaused?.Invoke(); } - private void OnConnectionReconnect() { - _ = HandleReconnected(); - } - - private async Task HandleReconnected() { + internal async void HandleReconnected() { try { // 打开 Session await SessionController.Reopen(); @@ -473,12 +485,12 @@ namespace LeanCloud.Realtime { OnResume?.Invoke(); } catch (Exception e) { LCLogger.Error(e); - await Connection.Close(); + Connection.UnRegister(this); // TODO 告知 OnClose?.Invoke(0, string.Empty); } } - + internal async Task GetOrQueryConversation(string convId) { if (ConversationDict.TryGetValue(convId, out LCIMConversation conversation)) { return conversation; diff --git a/Realtime/Realtime/LCRealtime.cs b/Realtime/Realtime/LCRealtime.cs index bd313b3..6f579a4 100644 --- a/Realtime/Realtime/LCRealtime.cs +++ b/Realtime/Realtime/LCRealtime.cs @@ -17,12 +17,19 @@ namespace LeanCloud.Realtime { if (appToConnections.TryGetValue(appId, out LCConnection connection)) { return connection; } - string connId = appId.Substring(0, 8).ToLower(); - connection = new LCConnection(connId); + connection = new LCConnection(appId); appToConnections[appId] = connection; return connection; } + /// + /// 移除 Connection + /// + /// + internal static void RemoveConnection(LCConnection connection) { + appToConnections.Remove(connection.id); + } + /// /// 主动断开所有 RTM 连接 /// From 772a00f52019585b4202296ec8f1e7956d288897 Mon Sep 17 00:00:00 2001 From: oneRain Date: Tue, 23 Jun 2020 16:03:24 +0800 Subject: [PATCH 3/9] =?UTF-8?q?chore:=20=E7=A7=BB=E9=99=A4=20Client=20?= =?UTF-8?q?=E5=AF=B9=20Connection=20=E7=9A=84=E4=BE=9D=E8=B5=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../Internal/Controller/LCIMController.cs | 2 +- .../Controller/LCIMConversationController.cs | 32 +++++++++---------- .../Controller/LCIMMessageController.cs | 12 +++---- .../Controller/LCIMSessionController.cs | 12 ++++--- Realtime/Realtime/LCIMClient.cs | 17 +--------- 5 files changed, 32 insertions(+), 43 deletions(-) diff --git a/Realtime/Realtime/Internal/Controller/LCIMController.cs b/Realtime/Realtime/Internal/Controller/LCIMController.cs index 78f134d..1a6d933 100644 --- a/Realtime/Realtime/Internal/Controller/LCIMController.cs +++ b/Realtime/Realtime/Internal/Controller/LCIMController.cs @@ -15,7 +15,7 @@ namespace LeanCloud.Realtime.Internal.Controller { protected LCConnection Connection { get { - return Client.Connection; + return LCRealtime.GetConnection(LCApplication.AppId); } } diff --git a/Realtime/Realtime/Internal/Controller/LCIMConversationController.cs b/Realtime/Realtime/Internal/Controller/LCIMConversationController.cs index 7a484bd..b2f288f 100644 --- a/Realtime/Realtime/Internal/Controller/LCIMConversationController.cs +++ b/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 allowedIds = response.ConvMessage.AllowedPids.ToList(); List 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 allowedIds = response.ConvMessage.AllowedPids.ToList(); List 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); } /// @@ -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); } /// @@ -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); } /// @@ -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(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(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 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 result = JsonConvert.DeserializeObject>(jsonObj.Data); if (result.TryGetValue(convId, out object obj)) { @@ -695,8 +695,8 @@ namespace LeanCloud.Realtime.Internal.Controller { /// 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); } diff --git a/Realtime/Realtime/Internal/Controller/LCIMMessageController.cs b/Realtime/Realtime/Internal/Controller/LCIMMessageController.cs index bfa0099..cec51cc 100644 --- a/Realtime/Realtime/Internal/Controller/LCIMMessageController.cs +++ b/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); } /// @@ -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); } /// @@ -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); } /// @@ -231,7 +231,7 @@ 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 diff --git a/Realtime/Realtime/Internal/Controller/LCIMSessionController.cs b/Realtime/Realtime/Internal/Controller/LCIMSessionController.cs index 73bce54..81e1861 100644 --- a/Realtime/Realtime/Internal/Controller/LCIMSessionController.cs +++ b/Realtime/Realtime/Internal/Controller/LCIMSessionController.cs @@ -20,13 +20,16 @@ namespace LeanCloud.Realtime.Internal.Controller { /// /// 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); } /// @@ -38,7 +41,7 @@ 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) { @@ -52,7 +55,8 @@ namespace LeanCloud.Realtime.Internal.Controller { /// internal async Task Close() { GenericCommand request = NewCommand(CommandType.Session, OpType.Close); - await Client.Connection.SendRequest(request); + await Connection.SendRequest(request); + Connection.UnRegister(Client); } /// @@ -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); } diff --git a/Realtime/Realtime/LCIMClient.cs b/Realtime/Realtime/LCIMClient.cs index e4c3acb..6402076 100644 --- a/Realtime/Realtime/LCIMClient.cs +++ b/Realtime/Realtime/LCIMClient.cs @@ -6,7 +6,6 @@ using System.Collections.ObjectModel; using LeanCloud.Storage; using LeanCloud.Realtime.Internal.Protocol; using LeanCloud.Realtime.Internal.Controller; -using LeanCloud.Realtime.Internal.Connection; namespace LeanCloud.Realtime { /// @@ -233,10 +232,6 @@ namespace LeanCloud.Realtime { get; private set; } - internal LCConnection Connection { - get; set; - } - internal LCIMSessionController SessionController { get; private set; } @@ -296,8 +291,6 @@ namespace LeanCloud.Realtime { ConversationController = new LCIMConversationController(this); MessageController = new LCIMMessageController(this); GoAwayController = new LCIMGoAwayController(this); - - Connection = LCRealtime.GetConnection(LCApplication.AppId); } /// @@ -306,15 +299,12 @@ namespace LeanCloud.Realtime { /// 是否强制登录 /// public async Task Open(bool force = true) { - await Connection.Connect(); try { // 打开 Session await SessionController.Open(force); - Connection.Register(this); } catch (Exception e) { LCLogger.Error(e); // 如果 session 阶段异常，则关闭连接 - Connection.UnRegister(this); throw e; } } @@ -326,7 +316,6 @@ namespace LeanCloud.Realtime { public async Task Close() { // 关闭 session await SessionController.Close(); - Connection.UnRegister(this); } /// @@ -449,9 +438,6 @@ namespace LeanCloud.Realtime { #endregion internal void HandleNotification(GenericCommand notification) { - if (notification.PeerId != Id) { - return; - } switch (notification.Cmd) { case CommandType.Session: SessionController.HandleNotification(notification); @@ -485,8 +471,7 @@ namespace LeanCloud.Realtime { OnResume?.Invoke(); } catch (Exception e) { LCLogger.Error(e); - Connection.UnRegister(this); - // TODO 告知 + // 重连成功，但 session/open 失败 OnClose?.Invoke(0, string.Empty); } } From 4d9769eeb7fdd233d9abbfeb7f64ec5b45c984ba Mon Sep 17 00:00:00 2001 From: oneRain Date: Tue, 23 Jun 2020 16:16:21 +0800 Subject: [PATCH 4/9] =?UTF-8?q?chore:=20=E7=AE=80=E5=8C=96=20Connection?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../Internal/Connection/LCConnection.cs | 22 +++---------------- 1 file changed, 3 insertions(+), 19 deletions(-) diff --git a/Realtime/Realtime/Internal/Connection/LCConnection.cs b/Realtime/Realtime/Internal/Connection/LCConnection.cs index 2e2cb8f..a2022b4 100644 --- a/Realtime/Realtime/Internal/Connection/LCConnection.cs +++ b/Realtime/Realtime/Internal/Connection/LCConnection.cs @@ -54,21 +54,6 @@ namespace LeanCloud.Realtime.Internal.Connection { /// private const string SUB_PROTOCOL = "lc.protobuf2.3"; - /// - /// 通知事件 - /// - internal Action OnNotification; - - /// - /// 断线事件 - /// - internal Action OnDisconnect; - - /// - /// 重连成功事件 - /// - internal Action OnReconnected; - internal string id; /// @@ -190,9 +175,6 @@ namespace LeanCloud.Realtime.Internal.Connection { /// internal async Task Close() { LCRealtime.RemoveConnection(this); - OnNotification = null; - OnDisconnect = null; - OnReconnected = null; heartBeat.Stop(); await ws.Close(); } @@ -277,7 +259,9 @@ namespace LeanCloud.Realtime.Internal.Connection { LCLogger.Debug("Reconnected"); ws.OnMessage = OnClientMessage; ws.OnClose = OnClientDisconnect; - OnReconnected?.Invoke(); + foreach (LCIMClient client in clients.Values) { + client.HandleReconnected(); + } break; } else { // 重置 Router，继续尝试重连 From d5879d777aa6141d11a116fb6286c613bccc1b78 Mon Sep 17 00:00:00 2001 From: oneRain Date: Tue, 23 Jun 2020 16:38:18 +0800 Subject: [PATCH 5/9] rename --- .../Internal/Connection/LCConnection.cs | 18 +++++++++--------- 1 file changed, 9 insertions(+), 9 deletions(-) diff --git a/Realtime/Realtime/Internal/Connection/LCConnection.cs b/Realtime/Realtime/Internal/Connection/LCConnection.cs index a2022b4..1d60bc7 100644 --- a/Realtime/Realtime/Internal/Connection/LCConnection.cs +++ b/Realtime/Realtime/Internal/Connection/LCConnection.cs @@ -72,7 +72,7 @@ namespace LeanCloud.Realtime.Internal.Connection { private State state; private Task connectTask; - private readonly Dictionary clients; + private readonly Dictionary idToClients; internal LCConnection(string id) { this.id = id; @@ -83,7 +83,7 @@ namespace LeanCloud.Realtime.Internal.Connection { OnMessage = OnClientMessage, OnClose = OnClientDisconnect }; - clients = new Dictionary(); + idToClients = new Dictionary(); state = State.None; } @@ -207,7 +207,7 @@ namespace LeanCloud.Realtime.Internal.Connection { heartBeat.Pong(); } else { // 通知 - if (clients.TryGetValue(command.PeerId, out LCIMClient client)) { + if (idToClients.TryGetValue(command.PeerId, out LCIMClient client)) { // 通知具体客户端 client.HandleNotification(command); } @@ -221,7 +221,7 @@ namespace LeanCloud.Realtime.Internal.Connection { private void OnClientDisconnect() { state = State.Closed; heartBeat.Stop(); - foreach (LCIMClient client in clients.Values) { + foreach (LCIMClient client in idToClients.Values) { client.HandleDisconnected(); } // 重连 @@ -231,7 +231,7 @@ namespace LeanCloud.Realtime.Internal.Connection { private void OnPingTimeout() { state = State.Closed; _ = ws.Close(); - foreach (LCIMClient client in clients.Values) { + foreach (LCIMClient client in idToClients.Values) { client.HandleDisconnected(); } // 重连 @@ -259,7 +259,7 @@ namespace LeanCloud.Realtime.Internal.Connection { LCLogger.Debug("Reconnected"); ws.OnMessage = OnClientMessage; ws.OnClose = OnClientDisconnect; - foreach (LCIMClient client in clients.Values) { + foreach (LCIMClient client in idToClients.Values) { client.HandleReconnected(); } break; @@ -280,12 +280,12 @@ namespace LeanCloud.Realtime.Internal.Connection { } internal void Register(LCIMClient client) { - clients[client.Id] = client; + idToClients[client.Id] = client; } internal void UnRegister(LCIMClient client) { - clients.Remove(client.Id); - if (clients.Count == 0) { + idToClients.Remove(client.Id); + if (idToClients.Count == 0) { _ = Close(); } } From b353bf5536d7372cdf7f2ff9fb65f0efc98f520e Mon Sep 17 00:00:00 2001 From: oneRain Date: Tue, 23 Jun 2020 18:37:09 +0800 Subject: [PATCH 6/9] =?UTF-8?q?chore:=20=E8=B0=83=E6=95=B4=20goaway=20?= =?UTF-8?q?=E5=A4=84=E7=90=86=E9=80=BB=E8=BE=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../Internal/Connection/LCConnection.cs | 3 +++ .../Controller/LCIMGoAwayController.cs | 19 ------------------- Realtime/Realtime/LCIMClient.cs | 8 -------- 3 files changed, 3 insertions(+), 27 deletions(-) delete mode 100644 Realtime/Realtime/Internal/Controller/LCIMGoAwayController.cs diff --git a/Realtime/Realtime/Internal/Connection/LCConnection.cs b/Realtime/Realtime/Internal/Connection/LCConnection.cs index 1d60bc7..2f11347 100644 --- a/Realtime/Realtime/Internal/Connection/LCConnection.cs +++ b/Realtime/Realtime/Internal/Connection/LCConnection.cs @@ -123,6 +123,7 @@ namespace LeanCloud.Realtime.Internal.Connection { /// /// internal async Task Reset() { + state = State.Closed; heartBeat?.Stop(); // 关闭就连接 await ws.Close(); @@ -205,6 +206,8 @@ namespace LeanCloud.Realtime.Internal.Connection { } else { if (command.Cmd == CommandType.Echo) { heartBeat.Pong(); + } else if (command.Cmd == CommandType.Goaway) { + _ = Reset(); } else { // 通知 if (idToClients.TryGetValue(command.PeerId, out LCIMClient client)) { diff --git a/Realtime/Realtime/Internal/Controller/LCIMGoAwayController.cs b/Realtime/Realtime/Internal/Controller/LCIMGoAwayController.cs deleted file mode 100644 index 35d4518..0000000 --- a/Realtime/Realtime/Internal/Controller/LCIMGoAwayController.cs +++ /dev/null @@ -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 void HandleNotification(GenericCommand notification) { - // 清空缓存，断开连接，等待重新连接 - _ = Connection.Reset(); - } - - #endregion - } -} diff --git a/Realtime/Realtime/LCIMClient.cs b/Realtime/Realtime/LCIMClient.cs index 6402076..6e08d3c 100644 --- a/Realtime/Realtime/LCIMClient.cs +++ b/Realtime/Realtime/LCIMClient.cs @@ -240,10 +240,6 @@ namespace LeanCloud.Realtime { get; private set; } - internal LCIMGoAwayController GoAwayController { - get; private set; - } - internal LCIMConversationController ConversationController { get; private set; } @@ -290,7 +286,6 @@ namespace LeanCloud.Realtime { SessionController = new LCIMSessionController(this); ConversationController = new LCIMConversationController(this); MessageController = new LCIMMessageController(this); - GoAwayController = new LCIMGoAwayController(this); } /// @@ -451,9 +446,6 @@ namespace LeanCloud.Realtime { case CommandType.Rcp: MessageController.HandleNotification(notification); break; - case CommandType.Goaway: - GoAwayController.HandleNotification(notification); - break; default: break; } From e30d528416495831fabe3115ddfbbb3d3d3bbaab Mon Sep 17 00:00:00 2001 From: oneRain Date: Wed, 24 Jun 2020 12:15:19 +0800 Subject: [PATCH 7/9] =?UTF-8?q?chore:=20=E6=94=AF=E6=8C=81=E4=B8=BB?= =?UTF-8?q?=E5=8A=A8=E6=96=AD=E5=BC=80=E5=92=8C=E9=87=8D=E8=BF=9E?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../Internal/Connection/LCConnection.cs | 81 ++++++++++--------- Realtime/Realtime/LCRealtime.cs | 8 +- Sample/RealtimeApp/Program.cs | 35 +++++++- 3 files changed, 79 insertions(+), 45 deletions(-) diff --git a/Realtime/Realtime/Internal/Connection/LCConnection.cs b/Realtime/Realtime/Internal/Connection/LCConnection.cs index 2f11347..db350b7 100644 --- a/Realtime/Realtime/Internal/Connection/LCConnection.cs +++ b/Realtime/Realtime/Internal/Connection/LCConnection.cs @@ -114,29 +114,11 @@ namespace LeanCloud.Realtime.Internal.Connection { heartBeat.Start(); state = State.Open; } catch (Exception e) { + state = State.Closed; throw e; } } - /// - /// 重置连接 - /// - /// - internal async Task Reset() { - state = State.Closed; - heartBeat?.Stop(); - // 关闭就连接 - await ws.Close(); - // 重新创建连接组件 - heartBeat = new LCHeartBeat(this, OnPingTimeout); - router = new LCRTMRouter(); - ws = new LCWebSocketClient { - OnMessage = OnClientMessage, - OnClose = OnClientDisconnect - }; - await Reconnect(); - } - /// /// 发送请求，会在收到应答后返回 /// @@ -171,15 +153,21 @@ namespace LeanCloud.Realtime.Internal.Connection { } /// - /// 关闭连接 + /// 断开连接 /// - /// - internal async Task Close() { - LCRealtime.RemoveConnection(this); + private void Disconnect() { + state = State.Closed; heartBeat.Stop(); - await ws.Close(); + _ = ws.Close(); + foreach (LCIMClient client in idToClients.Values) { + client.HandleDisconnected(); + } } + /// + /// 消息接收回调 + /// + /// private void OnClientMessage(byte[] bytes) { try { GenericCommand command = GenericCommand.Parser.ParseFrom(bytes); @@ -207,7 +195,7 @@ namespace LeanCloud.Realtime.Internal.Connection { if (command.Cmd == CommandType.Echo) { heartBeat.Pong(); } else if (command.Cmd == CommandType.Goaway) { - _ = Reset(); + Reset(); } else { // 通知 if (idToClients.TryGetValue(command.PeerId, out LCIMClient client)) { @@ -221,23 +209,35 @@ namespace LeanCloud.Realtime.Internal.Connection { } } + /// + /// 连接断开回调 + /// private void OnClientDisconnect() { - state = State.Closed; - heartBeat.Stop(); - foreach (LCIMClient client in idToClients.Values) { - client.HandleDisconnected(); - } - // 重连 + Disconnect(); _ = Reconnect(); } + /// + /// Pong 超时回调 + /// private void OnPingTimeout() { - state = State.Closed; - _ = ws.Close(); - foreach (LCIMClient client in idToClients.Values) { - client.HandleDisconnected(); - } - // 重连 + Disconnect(); + _ = Reconnect(); + } + + /// + /// 重置连接 + /// + /// + internal void Reset() { + Disconnect(); + // 重新创建连接组件 + heartBeat = new LCHeartBeat(this, OnPingTimeout); + router = new LCRTMRouter(); + ws = new LCWebSocketClient { + OnMessage = OnClientMessage, + OnClose = OnClientDisconnect + }; _ = Reconnect(); } @@ -289,7 +289,8 @@ namespace LeanCloud.Realtime.Internal.Connection { internal void UnRegister(LCIMClient client) { idToClients.Remove(client.Id); if (idToClients.Count == 0) { - _ = Close(); + Disconnect(); + LCRealtime.RemoveConnection(this); } } @@ -297,14 +298,14 @@ namespace LeanCloud.Realtime.Internal.Connection { /// 暂停连接 /// internal void Pause() { - + Disconnect(); } /// /// 恢复连接 /// internal void Resume() { - + _ = Reconnect(); } } } diff --git a/Realtime/Realtime/LCRealtime.cs b/Realtime/Realtime/LCRealtime.cs index 6f579a4..612650e 100644 --- a/Realtime/Realtime/LCRealtime.cs +++ b/Realtime/Realtime/LCRealtime.cs @@ -34,14 +34,18 @@ namespace LeanCloud.Realtime { /// 主动断开所有 RTM 连接 /// public static void Pause() { - + foreach (LCConnection connection in appToConnections.Values) { + connection.Pause(); + } } /// /// 主动恢复所有 RTM 连接 /// public static void Resume() { - + foreach (LCConnection connection in appToConnections.Values) { + connection.Resume(); + } } } } diff --git a/Sample/RealtimeApp/Program.cs b/Sample/RealtimeApp/Program.cs index 01ef348..1680362 100644 --- a/Sample/RealtimeApp/Program.cs +++ b/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}"); + } }); } From c358fd4de9263a95c1e7a583b759e6961c16b53e Mon Sep 17 00:00:00 2001 From: oneRain Date: Wed, 24 Jun 2020 12:21:16 +0800 Subject: [PATCH 8/9] =?UTF-8?q?chore:=20=E7=AE=80=E5=8C=96=E8=BF=9E?= =?UTF-8?q?=E6=8E=A5=E5=BC=82=E5=B8=B8=E5=A4=84=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../Internal/Connection/LCConnection.cs | 28 +++++++------------ 1 file changed, 10 insertions(+), 18 deletions(-) diff --git a/Realtime/Realtime/Internal/Connection/LCConnection.cs b/Realtime/Realtime/Internal/Connection/LCConnection.cs index db350b7..40312d7 100644 --- a/Realtime/Realtime/Internal/Connection/LCConnection.cs +++ b/Realtime/Realtime/Internal/Connection/LCConnection.cs @@ -77,11 +77,11 @@ namespace LeanCloud.Realtime.Internal.Connection { internal LCConnection(string id) { this.id = id; responses = new Dictionary>(); - heartBeat = new LCHeartBeat(this, OnPingTimeout); + heartBeat = new LCHeartBeat(this, OnDisconnect); router = new LCRTMRouter(); ws = new LCWebSocketClient { - OnMessage = OnClientMessage, - OnClose = OnClientDisconnect + OnMessage = OnMessage, + OnClose = OnDisconnect }; idToClients = new Dictionary(); state = State.None; @@ -168,7 +168,7 @@ namespace LeanCloud.Realtime.Internal.Connection { /// 消息接收回调 /// /// - private void OnClientMessage(byte[] bytes) { + private void OnMessage(byte[] bytes) { try { GenericCommand command = GenericCommand.Parser.ParseFrom(bytes); LCLogger.Debug($"{id} <= {FormatCommand(command)}"); @@ -212,15 +212,7 @@ namespace LeanCloud.Realtime.Internal.Connection { /// /// 连接断开回调 /// - private void OnClientDisconnect() { - Disconnect(); - _ = Reconnect(); - } - - /// - /// Pong 超时回调 - /// - private void OnPingTimeout() { + private void OnDisconnect() { Disconnect(); _ = Reconnect(); } @@ -232,11 +224,11 @@ namespace LeanCloud.Realtime.Internal.Connection { internal void Reset() { Disconnect(); // 重新创建连接组件 - heartBeat = new LCHeartBeat(this, OnPingTimeout); + heartBeat = new LCHeartBeat(this, OnDisconnect); router = new LCRTMRouter(); ws = new LCWebSocketClient { - OnMessage = OnClientMessage, - OnClose = OnClientDisconnect + OnMessage = OnMessage, + OnClose = OnDisconnect }; _ = Reconnect(); } @@ -260,8 +252,8 @@ namespace LeanCloud.Realtime.Internal.Connection { if (reconnectCount < MAX_RECONNECT_TIMES) { // 重连成功 LCLogger.Debug("Reconnected"); - ws.OnMessage = OnClientMessage; - ws.OnClose = OnClientDisconnect; + ws.OnMessage = OnMessage; + ws.OnClose = OnDisconnect; foreach (LCIMClient client in idToClients.Values) { client.HandleReconnected(); } From 4f38ce06158cd80559f5dcf6efa644ad8888bb9a Mon Sep 17 00:00:00 2001 From: oneRain Date: Wed, 24 Jun 2020 13:56:28 +0800 Subject: [PATCH 9/9] chore --- Realtime/Realtime-Unity/Realtime-Unity.csproj | 3 --- Realtime/Realtime/Internal/Connection/LCConnection.cs | 4 ++++ 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/Realtime/Realtime-Unity/Realtime-Unity.csproj b/Realtime/Realtime-Unity/Realtime-Unity.csproj index b900418..7e1a4a8 100644 --- a/Realtime/Realtime-Unity/Realtime-Unity.csproj +++ b/Realtime/Realtime-Unity/Realtime-Unity.csproj @@ -52,9 +52,6 @@ Internal\Controller\LCIMConversationController.cs - - Internal\Controller\LCIMGoAwayController.cs - Internal\Router\LCRTMServer.cs diff --git a/Realtime/Realtime/Internal/Connection/LCConnection.cs b/Realtime/Realtime/Internal/Connection/LCConnection.cs index 40312d7..3b797ac 100644 --- a/Realtime/Realtime/Internal/Connection/LCConnection.cs +++ b/Realtime/Realtime/Internal/Connection/LCConnection.cs @@ -70,8 +70,10 @@ namespace LeanCloud.Realtime.Internal.Connection { private LCWebSocketClient ws; private State state; + // 可以在 connecting 状态时拿到 Task，并在重连成功后继续操作 private Task connectTask; + // 共享这条连接的 IM Client private readonly Dictionary idToClients; internal LCConnection(string id) { @@ -193,8 +195,10 @@ namespace LeanCloud.Realtime.Internal.Connection { } } else { if (command.Cmd == CommandType.Echo) { + // 心跳应答 heartBeat.Pong(); } else if (command.Cmd == CommandType.Goaway) { + // 针对连接的消息 Reset(); } else { // 通知