Merge pull request #77 from onerain88/pingpong

* LCLiveQueryHeartBeat.cs:

LiveQuery/LiveQuery/Internal/LCLiveQueryConnection.cs

@@ -6,7 +6,7 @@ using Newtonsoft.Json;
 using LeanCloud.Realtime.Internal.Router;
 using LeanCloud.Realtime.Internal.WebSocket;
 using LeanCloud.Common;
-using LeanCloud.Storage;
+using LeanCloud.Realtime.Internal.Connection;
 
 namespace LeanCloud.LiveQuery.Internal {
     public class LCLiveQueryConnection {
@@ -63,7 +63,7 @@ namespace LeanCloud.LiveQuery.Internal {
         public LCLiveQueryConnection(string id) {
             this.id = id;
             responses = new Dictionary<int, TaskCompletionSource<Dictionary<string, object>>>();
-            heartBeat = new LCLiveQueryHeartBeat(this);
+            heartBeat = new LCLiveQueryHeartBeat(this, OnPingTimeout);
             router = new LCRTMRouter();
             client = new LCWebSocketClient {
                 OnMessage = OnClientMessage,
@@ -82,6 +82,8 @@ namespace LeanCloud.LiveQuery.Internal {
                     LCLogger.Debug($"Secondary Server");
                     await client.Connect(rtmServer.Secondary, SUB_PROTOCOL);
                 }
+                // 启动心跳
+                heartBeat.Start();
             } catch (Exception e) {
                 throw e;
             }
@@ -92,10 +94,11 @@ namespace LeanCloud.LiveQuery.Internal {
         /// </summary>
         /// <returns></returns>
         internal async Task Reset() {
+            heartBeat?.Stop();
             // 关闭就连接
             await client.Close();
             // 重新创建连接组件
-            heartBeat = new LCLiveQueryHeartBeat(this);
+            heartBeat = new LCLiveQueryHeartBeat(this, OnPingTimeout);
             router = new LCRTMRouter();
             client = new LCWebSocketClient {
                 OnMessage = OnClientMessage,
@@ -151,7 +154,6 @@ namespace LeanCloud.LiveQuery.Internal {
         }
 
         private void OnClientMessage(byte[] bytes) {
-            _ = heartBeat.Refresh(OnPingTimeout);
             try {
                 string json = Encoding.UTF8.GetString(bytes);
                 Dictionary<string, object> msg = JsonConvert.DeserializeObject<Dictionary<string, object>>(json,
@@ -177,8 +179,12 @@ namespace LeanCloud.LiveQuery.Internal {
                         LCLogger.Error($"No request for {requestIndex}");
                     }
                 } else {
-                    // 通知
+                    if (json == "{}") {
-                    OnNotification?.Invoke(msg);
+                        heartBeat.Pong();
+                    } else {
+                        // 通知
+                        OnNotification?.Invoke(msg);
+                    }
                 }
             } catch (Exception e) {
                 LCLogger.Error(e);
@@ -194,7 +200,9 @@ namespace LeanCloud.LiveQuery.Internal {
 
         private async void OnPingTimeout() {
             await client.Close();
-            OnClientDisconnect();
+            OnDisconnect?.Invoke();
+            // 重连
+            _ = Reconnect();
         }
 
         private async Task Reconnect() {

LiveQuery/LiveQuery/Internal/LCLiveQueryHeartBeat.cs

@@ -1,55 +1,27 @@
 using System;
 using System.Threading;
 using System.Threading.Tasks;
+using LeanCloud.Realtime.Internal.Connection;
 using System.Collections.Generic;
 
 namespace LeanCloud.LiveQuery.Internal {
     /// <summary>
     /// LiveQuery 心跳控制器
     /// </summary>
-    internal class LCLiveQueryHeartBeat {
+    internal class LCLiveQueryHeartBeat : LCHeartBeat {
-        private const int PING_INTERVAL = 5000;
-        private const int PONG_INTERVAL = 5000;
-
         private readonly LCLiveQueryConnection connection;
 
-        private CancellationTokenSource pingCTS;
+        internal LCLiveQueryHeartBeat(LCLiveQueryConnection connection,
-        private CancellationTokenSource pongCTS;
+            Action onTimeout) : base(onTimeout) {
-
-        internal LCLiveQueryHeartBeat(LCLiveQueryConnection connection) {
             this.connection = connection;
         }
 
-        internal async Task Refresh(Action onTimeout) {
+        protected override void SendPing() {
-            LCLogger.Debug("LiveQuery HeartBeat refresh");
+            try {
-            Stop();
+                _ = connection.SendText("{}");
+            } catch (Exception) {
 
-            pingCTS = new CancellationTokenSource();
-            Task delayTask = Task.Delay(PING_INTERVAL, pingCTS.Token);
-            await delayTask;
-            if (delayTask.IsCanceled) {
-                return;
             }
-            // 发送 Ping 包
-            LCLogger.Debug("Ping ~~~");
-            _ = connection.SendText("{}");
-
-            // 等待 Pong
-            pongCTS = new CancellationTokenSource();
-            Task timeoutTask = Task.Delay(PONG_INTERVAL, pongCTS.Token);
-            await timeoutTask;
-            if (timeoutTask.IsCanceled) {
-                return;
-            }
-
-            // 超时
-            LCLogger.Debug("Ping timeout");
-            onTimeout?.Invoke();
-        }
-
-        internal void Stop() {
-            pingCTS?.Cancel();
-            pongCTS?.Cancel();
         }
     }
 }

Realtime/Realtime/Internal/Connection/LCConnection.cs

@@ -6,13 +6,12 @@ using Google.Protobuf;
 using LeanCloud.Realtime.Internal.Router;
 using LeanCloud.Realtime.Internal.WebSocket;
 using LeanCloud.Realtime.Internal.Protocol;
-using LeanCloud.Storage;
 
 namespace LeanCloud.Realtime.Internal.Connection {
     /// <summary>
     /// 连接层，只与数据协议相关
     /// </summary>
-    internal class LCConnection {
+    public class LCConnection {
         /// <summary>
         /// 发送超时
         /// </summary>
@@ -28,11 +27,6 @@ namespace LeanCloud.Realtime.Internal.Connection {
         /// </summary>
         private const int RECONNECT_INTERVAL = 10000;
 
-        /// <summary>
-        /// 心跳间隔
-        /// </summary>
-        private const int HEART_BEAT_INTERVAL = 30000;
-
         /// <summary>
         /// 子协议
         /// </summary>
@@ -71,7 +65,7 @@ namespace LeanCloud.Realtime.Internal.Connection {
         internal LCConnection(string id) {
             this.id = id;
             responses = new Dictionary<int, TaskCompletionSource<GenericCommand>>();
-            heartBeat = new LCHeartBeat(this, HEART_BEAT_INTERVAL, HEART_BEAT_INTERVAL);
+            heartBeat = new LCHeartBeat(this, OnPingTimeout);
             router = new LCRTMRouter();
             client = new LCWebSocketClient {
                 OnMessage = OnClientMessage,
@@ -90,6 +84,8 @@ namespace LeanCloud.Realtime.Internal.Connection {
                     LCLogger.Debug($"Secondary Server");
                     await client.Connect(rtmServer.Secondary, SUB_PROTOCOL);
                 }
+                // 启动心跳
+                heartBeat.Start();
             } catch (Exception e) {
                 throw e;
             }
@@ -100,17 +96,18 @@ namespace LeanCloud.Realtime.Internal.Connection {
         /// </summary>
         /// <returns></returns>
         internal async Task Reset() {
+            heartBeat?.Stop();
             // 关闭就连接
             await client.Close();
             // 重新创建连接组件
-            heartBeat = new LCHeartBeat(this, HEART_BEAT_INTERVAL, HEART_BEAT_INTERVAL);
+            heartBeat = new LCHeartBeat(this, OnPingTimeout);
             router = new LCRTMRouter();
             client = new LCWebSocketClient {
                 OnMessage = OnClientMessage,
                 OnClose = OnClientDisconnect
             };
             await Reconnect();
-        }
+        }   
 
         /// <summary>
         /// 发送请求，会在收到应答后返回
@@ -158,7 +155,6 @@ namespace LeanCloud.Realtime.Internal.Connection {
         }
 
         private void OnClientMessage(byte[] bytes) {
-            _ = heartBeat.Refresh(OnPingTimeout);
             try {
                 GenericCommand command = GenericCommand.Parser.ParseFrom(bytes);
                 LCLogger.Debug($"{id} <= {FormatCommand(command)}");
@@ -182,8 +178,12 @@ namespace LeanCloud.Realtime.Internal.Connection {
                         LCLogger.Error($"No request for {requestIndex}");
                     }
                 } else {
-                    // 通知
+                    if (command.Cmd == CommandType.Echo) {
-                    OnNotification?.Invoke(command);
+                        heartBeat.Pong();
+                    } else {
+                        // 通知
+                        OnNotification?.Invoke(command);
+                    }
                 }
             } catch (Exception e) {
                 LCLogger.Error(e);
@@ -199,7 +199,9 @@ namespace LeanCloud.Realtime.Internal.Connection {
 
         private async void OnPingTimeout() {
             await client.Close();
-            OnClientDisconnect();
+            OnDisconnect?.Invoke();
+            // 重连
+            _ = Reconnect();
         }
 
         private async Task Reconnect() {

Realtime/Realtime/Internal/Connection/LCHeartBeat.cs

@@ -6,75 +6,98 @@ using LeanCloud.Realtime.Internal.Protocol;
 namespace LeanCloud.Realtime.Internal.Connection {
     /// <summary>
     /// 心跳控制器，由于 .Net Standard 2.0 不支持发送 ping frame，所以需要发送逻辑心跳
-    /// 1. 每次接收到消息后开始监听，如果在 pingInterval 时间内没有再次接收到消息，则发送 ping 请求；
+    /// 1. 每隔 180s 发送 ping 包
-    /// 2. 发送后等待 pongInterval 时间，如果在此时间内接收到了任何消息，则取消并重新开始监听 1；
+    /// 2. 接收到 pong 包刷新上次 pong 时间
-    /// 3. 如果没收到消息，则认为超时并回调，连接层接收回调后放弃当前连接，以断线逻辑处理
+    /// 3. 每隔 180s 检测 pong 包间隔，超过 360s 则认为断开
     /// </summary>
-    internal class LCHeartBeat {
+    public class LCHeartBeat {
+        private const int PING_INTERVAL = 5 * 1000;
+
         private readonly LCConnection connection;
 
-        /// <summary>
+        private readonly Action onTimeout;
-        /// ping 间隔
-        /// </summary>
-        private readonly int pingInterval;
 
-        /// <summary>
+        private CancellationTokenSource heartBeatCTS;
-        /// pong 间隔
-        /// </summary>
-        private readonly int pongInterval;
 
-        private CancellationTokenSource pingCTS;
+        private bool running = false;
-        private CancellationTokenSource pongCTS;
+
+        private DateTimeOffset lastPongTime;
+
+        public LCHeartBeat(Action onTimeout) {
+            this.onTimeout = onTimeout;
+        }
 
         internal LCHeartBeat(LCConnection connection,
-            int pingInterval,
+            Action onTimeout) : this(onTimeout) {
-            int pongInterval) {
             this.connection = connection;
-            this.pingInterval = pingInterval;
-            this.pongInterval = pongInterval;
         }
 
         /// <summary>
-        /// 更新心跳监听
+        /// 启动心跳
         /// </summary>
-        /// <returns></returns>
+        public void Start() {
-        internal async Task Refresh(Action onTimeout) {
+            running = true;
-            LCLogger.Debug("HeartBeat refresh");
+            heartBeatCTS = new CancellationTokenSource();
-            pingCTS?.Cancel();
+            StartPing();
-            pongCTS?.Cancel();
+            StartPong();
+        }
 
-            // 计时准备 ping
+        private async void StartPing() {
-            pingCTS = new CancellationTokenSource();
+            while (running) {
-            Task delayTask = Task.Delay(pingInterval, pingCTS.Token);
+                try {
-            await delayTask;
+                    await Task.Delay(PING_INTERVAL, heartBeatCTS.Token);
-            if (delayTask.IsCanceled) {
+                } catch (TaskCanceledException) {
-                return;
+                    return;
+                }
+                LCLogger.Debug("Ping ~~~");
+                SendPing();
             }
+        }
+
+        protected virtual void SendPing() {
             // 发送 ping 包
-            LCLogger.Debug("Ping ~~~");
             GenericCommand command = new GenericCommand {
                 Cmd = CommandType.Echo,
                 AppId = LCApplication.AppId,
                 PeerId = connection.id
             };
-            _ = connection.SendRequest(command);
+            try {
-            pongCTS = new CancellationTokenSource();
+                _ = connection.SendCommand(command);
-            Task timeoutTask = Task.Delay(pongInterval, pongCTS.Token);
+            } catch (Exception e) {
-            await timeoutTask;
+                LCLogger.Error(e.Message);
-            if (timeoutTask.IsCanceled) {
-                return;
             }
-            // timeout
+        }
-            LCLogger.Error("Ping timeout");
+
-            onTimeout.Invoke();
+        private async void StartPong() {
+            lastPongTime = DateTimeOffset.Now;
+            while (running) {
+                try {
+                    await Task.Delay(PING_INTERVAL / 2, heartBeatCTS.Token);
+                } catch (TaskCanceledException) {
+                    return;
+                }
+                // 检查 pong 包时间
+                TimeSpan interval = DateTimeOffset.Now - lastPongTime;
+                if (interval.TotalMilliseconds > PING_INTERVAL * 2) {
+                    // 断线
+                    Stop();
+                    onTimeout.Invoke();
+                }
+            }
+        }
+
+        public void Pong() {
+            LCLogger.Debug("Pong ~~~");
+            // 刷新最近 pong 时间戳
+            lastPongTime = DateTimeOffset.Now;
         }
 
         /// <summary>
         /// 停止心跳监听
         /// </summary>
-        internal void Stop() {
+        public void Stop() {
-            pingCTS?.Cancel();
+            running = false;
-            pongCTS?.Cancel();
+            heartBeatCTS.Cancel();
         }
     }
 }