You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by di...@apache.org on 2022/04/19 02:38:26 UTC
[rocketmq-mqtt] branch main updated: [ISSUE #24] recalculate timeout check
This is an automated email from the ASF dual-hosted git repository.
dinglei pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/rocketmq-mqtt.git
The following commit(s) were added to refs/heads/main by this push:
new 5380490 [ISSUE #24] recalculate timeout check
5380490 is described below
commit 538049033b151c87c7d797cc51864f4805851c0a
Author: pingww <pi...@gmail.com>
AuthorDate: Tue Apr 19 10:38:22 2022 +0800
[ISSUE #24] recalculate timeout check
* [ISSUE #24] recalculate timeout check
* [ISSUE #24] uc , recalculate keepalive
* [ISSUE #24] uc , recalculate keepalive
---
.../rocketmq/mqtt/cs/channel/DefaultChannelManager.java | 9 +++++++--
.../mqtt/cs/test/channel/TestDefaultChannelManager.java | 11 +++++++++++
2 files changed, 18 insertions(+), 2 deletions(-)
diff --git a/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/channel/DefaultChannelManager.java b/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/channel/DefaultChannelManager.java
index ee2e8b8..f889f8a 100644
--- a/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/channel/DefaultChannelManager.java
+++ b/mqtt-cs/src/main/java/org/apache/rocketmq/mqtt/cs/channel/DefaultChannelManager.java
@@ -94,8 +94,13 @@ public class DefaultChannelManager implements ChannelManager {
closeConnect(channel, ChannelCloseFrom.SERVER, "No Heart");
} else {
int keepAliveTimeSeconds = ChannelInfo.getKeepLive(channel);
- hashedWheelTimer.newTimeout(timeout.task(), (long)Math.ceil(keepAliveTimeSeconds * 1.5 + 1),
- TimeUnit.SECONDS);
+ long lastTouchTime = ChannelInfo.getLastTouch(channel);
+ long heartWindow = (long) Math.ceil(keepAliveTimeSeconds * 1.5 + 1) * 1000L;
+ long delay = Math.min(heartWindow, heartWindow - (System.currentTimeMillis() - lastTouchTime));
+ if (delay <= 0) {
+ delay = heartWindow;
+ }
+ hashedWheelTimer.newTimeout(timeout.task(), delay, TimeUnit.MILLISECONDS);
}
} catch (Exception e) {
logger.error("Exception when doPing: ", e);
diff --git a/mqtt-cs/src/test/java/org/apache/rocketmq/mqtt/cs/test/channel/TestDefaultChannelManager.java b/mqtt-cs/src/test/java/org/apache/rocketmq/mqtt/cs/test/channel/TestDefaultChannelManager.java
index f854652..031e8cf 100644
--- a/mqtt-cs/src/test/java/org/apache/rocketmq/mqtt/cs/test/channel/TestDefaultChannelManager.java
+++ b/mqtt-cs/src/test/java/org/apache/rocketmq/mqtt/cs/test/channel/TestDefaultChannelManager.java
@@ -94,6 +94,17 @@ public class TestDefaultChannelManager {
verify(retryDriver).unloadSession(Mockito.any());
}
+ @Test
+ public void testKeepLive() throws InterruptedException {
+ ChannelInfo.setClientId(channel, clientId);
+ defaultChannelManager.addChannel(channel);
+ ChannelInfo.setKeepLive(channel, 1);
+ Thread.sleep(1000);
+ Assert.assertFalse(0 == defaultChannelManager.totalConn());
+ Thread.sleep(4000);
+ Assert.assertTrue(0 == defaultChannelManager.totalConn());
+ }
+
@Test
public void testCloseConnectNullClientId() {
defaultChannelManager.closeConnect(channel, ChannelCloseFrom.CLIENT, "ForTest");