Fix force ping request logic

gojek · Jul 18, 2023 · b7da0c7 · b7da0c7
1 parent 7ae7f5b
commit b7da0c7
Show file tree

Hide file tree

Showing 12 changed files with 112 additions and 36 deletions.
diff --git a/app/src/main/java/com/gojek/courier/app/ui/MainActivity.kt b/app/src/main/java/com/gojek/courier/app/ui/MainActivity.kt
@@ -153,7 +153,7 @@ class MainActivity : AppCompatActivity() {
                 incomingMessagesCleanupIntervalSecs = 10,
                 maxInflightMessagesLimit = 1000,
             ),
-            pingSender = WorkPingSenderFactory.createMqttPingSender(applicationContext, WorkManagerPingSenderConfig())
+            pingSender = WorkPingSenderFactory.createMqttPingSender(applicationContext, WorkManagerPingSenderConfig(sendForcePing = true))
         )
         mqttClient = MqttClientFactory.create(this, mqttConfig)
         mqttClient.addEventHandler(eventHandler)

diff --git a/paho/src/main/java/org/eclipse/paho/client/mqttv3/MqttAsyncClient.java b/paho/src/main/java/org/eclipse/paho/client/mqttv3/MqttAsyncClient.java
@@ -872,7 +872,7 @@ public IMqttToken checkPing(Object userContext, IMqttActionListener callback) th
 		// @TRACE 117=>
 		logger.d(TAG, "checking for ping");
 
-		token = comms.checkForActivity();
+		token = comms.checkForActivity(false);
 		// @TRACE 118=<
 
 		return token;

diff --git a/paho/src/main/java/org/eclipse/paho/client/mqttv3/TimerPingSender.java b/paho/src/main/java/org/eclipse/paho/client/mqttv3/TimerPingSender.java
@@ -86,7 +86,7 @@ public void run()
 		{
 			// @Trace 660=Check schedule at {0}
 			logger.d(TAG, "in ping timer task run function");
-			comms.checkForActivity();
+			comms.checkForActivity(false);
 		}
 	}
 }
diff --git a/paho/src/main/java/org/eclipse/paho/client/mqttv3/internal/ClientComms.java b/paho/src/main/java/org/eclipse/paho/client/mqttv3/internal/ClientComms.java
@@ -758,10 +758,10 @@ public void run() {
     /*
      * Check and send a ping if needed and check for ping timeout. Need to send a ping if nothing has been sent or received in the last keepalive interval.
      */
-    public MqttToken checkForActivity() {
+    public MqttToken checkForActivity(Boolean forcePing) {
         MqttToken token = null;
         try {
-            token = clientState.checkForActivity();
+            token = clientState.checkForActivity(forcePing);
         } catch (MqttException e) {
             handleRunException(e);
         } catch (Exception e) {

diff --git a/paho/src/main/java/org/eclipse/paho/client/mqttv3/internal/ClientState.java b/paho/src/main/java/org/eclipse/paho/client/mqttv3/internal/ClientState.java
@@ -649,7 +649,7 @@ protected void undo(MqttPublish message) throws MqttPersistenceException
 	 *
 	 * @return token of ping command, null if no ping command has been sent.
 	 */
-	public MqttToken checkForActivity() throws MqttException
+	public MqttToken checkForActivity(Boolean forcePing) throws MqttException
 	{
 		final String methodName = "checkForActivity";
 
@@ -672,7 +672,7 @@ public MqttToken checkForActivity() throws MqttException
 					long lastActivity = lastInboundActivity;
 
 					// Is a ping required?
-					if (time - lastActivity + keepAliveMargin >= this.keepAlive)
+					if (forcePing || (time - lastActivity + keepAliveMargin >= this.keepAlive))
 					{
 
 						// @TRACE 620=ping needed. keepAlive={0} lastOutboundActivity={1} lastInboundActivity={2}

diff --git a/pingsender/alarm-pingsender/src/main/java/com/gojek/alarm/pingsender/AlarmPingSender.kt b/pingsender/alarm-pingsender/src/main/java/com/gojek/alarm/pingsender/AlarmPingSender.kt
@@ -201,11 +201,7 @@ internal class AlarmPingSender(
                 serverUri = comms.client.serverURI
             }
             pingSenderEvents.mqttPingInitiated(serverUri, comms.keepAlive.fromMillisToSeconds())
-            val token = if (alarmPingSenderConfig.sendForcePing) {
-                comms.sendPingRequest()
-            } else {
-                comms.checkForActivity()
-            }
+            val token = comms.checkForActivity(alarmPingSenderConfig.sendForcePing)
 
             // No ping has been sent.
             if (token == null) {

diff --git a/pingsender/timer-pingsender/src/main/java/com/gojek/timer/pingsender/TimerPingSender.kt b/pingsender/timer-pingsender/src/main/java/com/gojek/timer/pingsender/TimerPingSender.kt
@@ -72,11 +72,7 @@ internal class TimerPingSender(
             val serverUri = comms.client?.serverURI ?: ""
             val keepAliveMillis = comms.keepAlive
             pingSenderEvents.mqttPingInitiated(comms.client.serverURI, keepAliveMillis.fromMillisToSeconds())
-            val token = if (pingSenderConfig.sendForcePing) {
-                comms.sendPingRequest()
-            } else {
-                comms.checkForActivity()
-            }
+            val token = comms.checkForActivity(pingSenderConfig.sendForcePing)
             if (token == null) {
                 logger.d(TAG, "Mqtt Ping Token null")
                 pingSenderEvents.pingMqttTokenNull(serverUri, keepAliveMillis.fromMillisToSeconds())

diff --git a/pingsender/timer-pingsender/src/test/java/com/gojek/timer/pingsender/TimerPingSenderTest.kt b/pingsender/timer-pingsender/src/test/java/com/gojek/timer/pingsender/TimerPingSenderTest.kt
@@ -73,10 +73,11 @@ class TimerPingSenderTest {
         val mqttClient = mock<IMqttAsyncClient>()
         val testUri = "test-uri"
         val keepaliveMillis = 30000L
+        whenever(pingSenderConfig.sendForcePing).thenReturn(false)
         whenever(comms.client).thenReturn(mqttClient)
         whenever(mqttClient.serverURI).thenReturn(testUri)
         whenever(comms.keepAlive).thenReturn(keepaliveMillis)
-        whenever(comms.checkForActivity()).thenReturn(null)
+        whenever(comms.checkForActivity(false)).thenReturn(null)
 
         pingSender.PingTask().run()
 
@@ -92,10 +93,36 @@ class TimerPingSenderTest {
         val keepaliveMillis = 30000L
         val startTime = TimeUnit.MILLISECONDS.toNanos(100)
         val endTime = TimeUnit.MILLISECONDS.toNanos(110)
+        whenever(pingSenderConfig.sendForcePing).thenReturn(false)
         whenever(comms.client).thenReturn(mqttClient)
         whenever(mqttClient.serverURI).thenReturn(testUri)
         whenever(comms.keepAlive).thenReturn(keepaliveMillis)
-        whenever(comms.checkForActivity()).thenReturn(mqttToken)
+        whenever(comms.checkForActivity(false)).thenReturn(mqttToken)
+        whenever(clock.nanoTime()).thenReturn(startTime, endTime)
+
+        pingSender.PingTask().run()
+
+        verify(pingSenderEvents).mqttPingInitiated(testUri, keepaliveMillis / 1000)
+
+        val argumentCaptor = argumentCaptor<IMqttActionListener>()
+        verify(mqttToken).actionCallback = argumentCaptor.capture()
+        argumentCaptor.lastValue.onSuccess(mqttToken)
+        verify(pingSenderEvents).pingEventSuccess(testUri, 10, keepaliveMillis / 1000)
+    }
+
+    @Test
+    fun `test sendPing when ping can be sent successfully with sendForcePing=true`() {
+        val mqttClient = mock<IMqttAsyncClient>()
+        val mqttToken = mock<MqttToken>()
+        val testUri = "test-uri"
+        val keepaliveMillis = 30000L
+        val startTime = TimeUnit.MILLISECONDS.toNanos(100)
+        val endTime = TimeUnit.MILLISECONDS.toNanos(110)
+        whenever(pingSenderConfig.sendForcePing).thenReturn(true)
+        whenever(comms.client).thenReturn(mqttClient)
+        whenever(mqttClient.serverURI).thenReturn(testUri)
+        whenever(comms.keepAlive).thenReturn(keepaliveMillis)
+        whenever(comms.checkForActivity(true)).thenReturn(mqttToken)
         whenever(clock.nanoTime()).thenReturn(startTime, endTime)
 
         pingSender.PingTask().run()
@@ -141,10 +168,11 @@ class TimerPingSenderTest {
         val keepaliveMillis = 30000L
         val startTime = TimeUnit.MILLISECONDS.toNanos(100)
         val endTime = TimeUnit.MILLISECONDS.toNanos(110)
+        whenever(pingSenderConfig.sendForcePing).thenReturn(false)
         whenever(comms.client).thenReturn(mqttClient)
         whenever(mqttClient.serverURI).thenReturn(testUri)
         whenever(comms.keepAlive).thenReturn(keepaliveMillis)
-        whenever(comms.checkForActivity()).thenReturn(mqttToken)
+        whenever(comms.checkForActivity(false)).thenReturn(mqttToken)
         whenever(clock.nanoTime()).thenReturn(startTime, endTime)
 
         pingSender.PingTask().run()

diff --git a/...-2.6.0-pingsender/src/main/java/com/gojek/workmanager/pingsender/WorkManagerPingSender.kt b/...-2.6.0-pingsender/src/main/java/com/gojek/workmanager/pingsender/WorkManagerPingSender.kt
@@ -54,11 +54,7 @@ internal class WorkManagerPingSender(
         val keepAliveMillis = comms.keepAlive
         pingSenderEvents.mqttPingInitiated(serverUri, keepAliveMillis.fromMillisToSeconds())
 
-        val token = if (pingSenderConfig.sendForcePing) {
-            comms.sendPingRequest()
-        } else {
-            comms.checkForActivity()
-        }
+        val token = comms.checkForActivity(pingSenderConfig.sendForcePing)
         if (token == null) {
             logger.d(TAG, "Mqtt Ping Token null")
             pingSenderEvents.pingMqttTokenNull(serverUri, keepAliveMillis.fromMillisToSeconds())

diff --git a/....0-pingsender/src/test/java/com/gojek/workmanager/pingsender/WorkManagerPingSenderTest.kt b/....0-pingsender/src/test/java/com/gojek/workmanager/pingsender/WorkManagerPingSenderTest.kt
@@ -61,10 +61,11 @@ class WorkManagerPingSenderTest {
         val mqttClient = mock<IMqttAsyncClient>()
         val testUri = "test-uri"
         val keepaliveMillis = 30000L
+        whenever(pingSenderConfig.sendForcePing).thenReturn(false)
         whenever(comms.client).thenReturn(mqttClient)
         whenever(mqttClient.serverURI).thenReturn(testUri)
         whenever(comms.keepAlive).thenReturn(keepaliveMillis)
-        whenever(comms.checkForActivity()).thenReturn(null)
+        whenever(comms.checkForActivity(false)).thenReturn(null)
 
         pingSender.sendPing {
             // do nothing
@@ -82,10 +83,40 @@ class WorkManagerPingSenderTest {
         val keepaliveMillis = 30000L
         val startTime = TimeUnit.MILLISECONDS.toNanos(100)
         val endTime = TimeUnit.MILLISECONDS.toNanos(110)
+        whenever(pingSenderConfig.sendForcePing).thenReturn(false)
         whenever(comms.client).thenReturn(mqttClient)
         whenever(mqttClient.serverURI).thenReturn(testUri)
         whenever(comms.keepAlive).thenReturn(keepaliveMillis)
-        whenever(comms.checkForActivity()).thenReturn(mqttToken)
+        whenever(comms.checkForActivity(false)).thenReturn(mqttToken)
+        whenever(clock.nanoTime()).thenReturn(startTime, endTime)
+
+        var success: Boolean? = null
+        pingSender.sendPing {
+            success = it
+        }
+
+        verify(pingSenderEvents).mqttPingInitiated(testUri, keepaliveMillis / 1000)
+
+        val argumentCaptor = argumentCaptor<IMqttActionListener>()
+        verify(mqttToken).actionCallback = argumentCaptor.capture()
+        argumentCaptor.lastValue.onSuccess(mqttToken)
+        assertTrue(success!!)
+        verify(pingSenderEvents).pingEventSuccess(testUri, 10, keepaliveMillis / 1000)
+    }
+
+    @Test
+    fun `test sendPing when ping can be sent successfully with sendForcePing=true`() {
+        val mqttClient = mock<IMqttAsyncClient>()
+        val mqttToken = mock<MqttToken>()
+        val testUri = "test-uri"
+        val keepaliveMillis = 30000L
+        val startTime = TimeUnit.MILLISECONDS.toNanos(100)
+        val endTime = TimeUnit.MILLISECONDS.toNanos(110)
+        whenever(pingSenderConfig.sendForcePing).thenReturn(true)
+        whenever(comms.client).thenReturn(mqttClient)
+        whenever(mqttClient.serverURI).thenReturn(testUri)
+        whenever(comms.keepAlive).thenReturn(keepaliveMillis)
+        whenever(comms.checkForActivity(true)).thenReturn(mqttToken)
         whenever(clock.nanoTime()).thenReturn(startTime, endTime)
 
         var success: Boolean? = null
@@ -139,10 +170,11 @@ class WorkManagerPingSenderTest {
         val keepaliveMillis = 30000L
         val startTime = TimeUnit.MILLISECONDS.toNanos(100)
         val endTime = TimeUnit.MILLISECONDS.toNanos(110)
+        whenever(pingSenderConfig.sendForcePing).thenReturn(false)
         whenever(comms.client).thenReturn(mqttClient)
         whenever(mqttClient.serverURI).thenReturn(testUri)
         whenever(comms.keepAlive).thenReturn(keepaliveMillis)
-        whenever(comms.checkForActivity()).thenReturn(mqttToken)
+        whenever(comms.checkForActivity(false)).thenReturn(mqttToken)
         whenever(clock.nanoTime()).thenReturn(startTime, endTime)
 
         var success: Boolean? = null

diff --git a/...anager-pingsender/src/main/java/com/gojek/workmanager/pingsender/WorkManagerPingSender.kt b/...anager-pingsender/src/main/java/com/gojek/workmanager/pingsender/WorkManagerPingSender.kt
@@ -54,11 +54,7 @@ internal class WorkManagerPingSender(
         val keepAliveMillis = comms.keepAlive
         pingSenderEvents.mqttPingInitiated(serverUri, keepAliveMillis.fromMillisToSeconds())
 
-        val token = if (pingSenderConfig.sendForcePing) {
-            comms.sendPingRequest()
-        } else {
-            comms.checkForActivity()
-        }
+        val token = comms.checkForActivity(pingSenderConfig.sendForcePing)
         if (token == null) {
             logger.d(TAG, "Mqtt Ping Token null")
             pingSenderEvents.pingMqttTokenNull(serverUri, keepAliveMillis.fromMillisToSeconds())

diff --git a/...er-pingsender/src/test/java/com/gojek/workmanager/pingsender/WorkManagerPingSenderTest.kt b/...er-pingsender/src/test/java/com/gojek/workmanager/pingsender/WorkManagerPingSenderTest.kt
@@ -61,10 +61,11 @@ class WorkManagerPingSenderTest {
         val mqttClient = mock<IMqttAsyncClient>()
         val testUri = "test-uri"
         val keepaliveMillis = 30000L
+        whenever(pingSenderConfig.sendForcePing).thenReturn(false)
         whenever(comms.client).thenReturn(mqttClient)
         whenever(mqttClient.serverURI).thenReturn(testUri)
         whenever(comms.keepAlive).thenReturn(keepaliveMillis)
-        whenever(comms.checkForActivity()).thenReturn(null)
+        whenever(comms.checkForActivity(false)).thenReturn(null)
 
         pingSender.sendPing {
             // do nothing
@@ -82,10 +83,40 @@ class WorkManagerPingSenderTest {
         val keepaliveMillis = 30000L
         val startTime = TimeUnit.MILLISECONDS.toNanos(100)
         val endTime = TimeUnit.MILLISECONDS.toNanos(110)
+        whenever(pingSenderConfig.sendForcePing).thenReturn(false)
         whenever(comms.client).thenReturn(mqttClient)
         whenever(mqttClient.serverURI).thenReturn(testUri)
         whenever(comms.keepAlive).thenReturn(keepaliveMillis)
-        whenever(comms.checkForActivity()).thenReturn(mqttToken)
+        whenever(comms.checkForActivity(false)).thenReturn(mqttToken)
+        whenever(clock.nanoTime()).thenReturn(startTime, endTime)
+
+        var success: Boolean? = null
+        pingSender.sendPing {
+            success = it
+        }
+
+        verify(pingSenderEvents).mqttPingInitiated(testUri, keepaliveMillis / 1000)
+
+        val argumentCaptor = argumentCaptor<IMqttActionListener>()
+        verify(mqttToken).actionCallback = argumentCaptor.capture()
+        argumentCaptor.lastValue.onSuccess(mqttToken)
+        assertTrue(success!!)
+        verify(pingSenderEvents).pingEventSuccess(testUri, 10, keepaliveMillis / 1000)
+    }
+
+    @Test
+    fun `test sendPing when ping can be sent successfully with sendForcePing=true`() {
+        val mqttClient = mock<IMqttAsyncClient>()
+        val mqttToken = mock<MqttToken>()
+        val testUri = "test-uri"
+        val keepaliveMillis = 30000L
+        val startTime = TimeUnit.MILLISECONDS.toNanos(100)
+        val endTime = TimeUnit.MILLISECONDS.toNanos(110)
+        whenever(pingSenderConfig.sendForcePing).thenReturn(true)
+        whenever(comms.client).thenReturn(mqttClient)
+        whenever(mqttClient.serverURI).thenReturn(testUri)
+        whenever(comms.keepAlive).thenReturn(keepaliveMillis)
+        whenever(comms.checkForActivity(true)).thenReturn(mqttToken)
         whenever(clock.nanoTime()).thenReturn(startTime, endTime)
 
         var success: Boolean? = null
@@ -139,10 +170,11 @@ class WorkManagerPingSenderTest {
         val keepaliveMillis = 30000L
         val startTime = TimeUnit.MILLISECONDS.toNanos(100)
         val endTime = TimeUnit.MILLISECONDS.toNanos(110)
+        whenever(pingSenderConfig.sendForcePing).thenReturn(false)
         whenever(comms.client).thenReturn(mqttClient)
         whenever(mqttClient.serverURI).thenReturn(testUri)
         whenever(comms.keepAlive).thenReturn(keepaliveMillis)
-        whenever(comms.checkForActivity()).thenReturn(mqttToken)
+        whenever(comms.checkForActivity(false)).thenReturn(mqttToken)
         whenever(clock.nanoTime()).thenReturn(startTime, endTime)
 
         var success: Boolean? = null