Merge pull request #4 from ShareChat/sync-main

Update courier lib to gojek courier lib latest version

.github/workflows/publish.yml

@@ -34,6 +34,7 @@ jobs:
           :timer-pingsender:publishReleasePublicationToSonatypeRepository
           :adaptive-keep-alive:publishReleasePublicationToSonatypeRepository
           :network-tracker:publishReleasePublicationToSonatypeRepository
+          :courier-message-adapter-text:publishReleasePublicationToSonatypeRepository
           :courier-message-adapter-gson:publishReleasePublicationToSonatypeRepository
           :courier-message-adapter-moshi:publishReleasePublicationToSonatypeRepository
           :courier-message-adapter-protobuf:publishReleasePublicationToSonatypeRepository

adaptive-keep-alive/build.gradle.kts

@@ -46,6 +46,7 @@ dependencies {
     implementation(project(":mqtt-pingsender"))
     implementation(project(":network-tracker"))
 
+    testImplementation(deps.android.test.mockitoCore)
     testImplementation(deps.android.test.kotlinTestJunit)
 }
 

adaptive-keep-alive/src/main/java/com/gojek/keepalive/AdaptiveKeepAliveStateHandler.kt

@@ -98,7 +98,7 @@ internal class AdaptiveKeepAliveStateHandler(
             )
         } else {
             val currentUpperBound = keepAlive.keepAliveMinutes - 1
-            if (state.lastSuccessfulKA == currentUpperBound) {
+            if (state.lastSuccessfulKA >= currentUpperBound) {
                 state = state.copy(
                     currentUpperBound = currentUpperBound,
                     isOptimalKeepAlive = true,
@@ -191,8 +191,7 @@ internal class AdaptiveKeepAliveStateHandler(
             keepAlive.keepAliveMinutes == state.currentKA
     }
 
-    @VisibleForTesting
-    internal fun resetState() {
+    fun resetState() {
         state = state.copy(
             lastSuccessfulKA = state.lowerBound - state.step,
             isOptimalKeepAlive = false,
@@ -201,7 +200,8 @@ internal class AdaptiveKeepAliveStateHandler(
             currentKA = -1,
             currentKAFailureCount = 0,
             probeCount = 0,
-            convergenceTime = 0
+            convergenceTime = 0,
+            optimalKAFailureCount = 0
         )
     }
 

adaptive-keep-alive/src/main/java/com/gojek/keepalive/OptimalKeepAliveCalculator.kt

@@ -20,11 +20,9 @@ internal class OptimalKeepAliveCalculator(
         object : NetworkStateListener {
             override fun onStateChanged(activeNetworkState: NetworkState) {
                 synchronized(this) {
-                    if (activeNetworkState.isConnected) {
-                        val networkType = networkUtils.getNetworkType(activeNetworkState.netInfo)
-                        val networkName = networkUtils.getNetworkName(activeNetworkState.netInfo)
-                        onNetworkStateChanged(networkType, networkName)
-                    }
+                    val networkType = networkUtils.getNetworkType(activeNetworkState.netInfo)
+                    val networkName = networkUtils.getNetworkName(activeNetworkState.netInfo)
+                    onNetworkStateChanged(networkType, networkName)
                 }
             }
         }
@@ -108,6 +106,7 @@ internal class OptimalKeepAliveCalculator(
             stateHandler.updateOptimalKeepAliveFailureState()
             if (stateHandler.isOptimalKeepAliveFailureLimitExceeded()) {
                 stateHandler.removeStateFromPersistence()
+                stateHandler.resetState()
             }
         }
     }

adaptive-keep-alive/src/test/java/com/gojek/keepalive/OptimalKeepAliveCalculatorTest.kt

@@ -39,10 +39,9 @@ class OptimalKeepAliveCalculatorTest {
     }
 
     @Test
-    fun `test onStateChanged should notify state handler when network is connected`() {
+    fun `test onStateChanged should notify state handler`() {
         val networkState = mock<NetworkState>()
         val netInfo = mock<NetworkInfo>()
-        whenever(networkState.isConnected).thenReturn(true)
         whenever(networkState.netInfo).thenReturn(netInfo)
         val networkType = 1
         val networkName = "test-network"
@@ -52,23 +51,12 @@ class OptimalKeepAliveCalculatorTest {
         optimalKeepAliveCalculator.networkStateListener.onStateChanged(networkState)
 
         verify(stateHandler).onNetworkChanged(networkType, networkName)
-        verify(networkState).isConnected
         verify(networkState, times(2)).netInfo
         verify(networkUtils).getNetworkType(netInfo)
         verify(networkUtils).getNetworkName(netInfo)
         verifyNoMoreInteractions(networkState)
     }
 
-    @Test
-    fun `test onStateChanged should not notify state handler when network is not connected`() {
-        val networkState = mock<NetworkState>()
-        whenever(networkState.isConnected).thenReturn(false)
-
-        optimalKeepAliveCalculator.networkStateListener.onStateChanged(networkState)
-
-        verify(networkState).isConnected
-    }
-
     @Test
     fun `test getUnderTrialKeepAlive when optimal keep alive is already found`() {
         val optimalKeepAlive = mock<KeepAlive>()
@@ -271,6 +259,7 @@ class OptimalKeepAliveCalculatorTest {
         verify(stateHandler).updateOptimalKeepAliveFailureState()
         verify(stateHandler).isOptimalKeepAliveFailureLimitExceeded()
         verify(stateHandler).removeStateFromPersistence()
+        verify(stateHandler).resetState()
     }
 
     @Test

app/src/main/java/com/gojek/courier/app/ui/MainActivity.kt

@@ -92,31 +92,29 @@ class MainActivity : AppCompatActivity() {
     }
 
     private fun connectMqtt(clientId: String, username: String, password: String, ip: String, port: Int) {
-        val connectOptions = MqttConnectOptions(
-            serverUris = listOf(ServerUri(ip, port, if (port == 443) "ssl" else "tcp")),
-            clientId = clientId,
-            username = username,
-            keepAlive = KeepAlive(
-                timeSeconds = 30
-            ),
-            isCleanSession = false,
-            password = password
-        )
+        val connectOptions = MqttConnectOptions.Builder()
+            .serverUris(listOf(ServerUri(ip, port, if (port == 443) "ssl" else "tcp")))
+            .clientId(clientId)
+            .userName(username)
+            .password(password)
+            .cleanSession(false)
+            .keepAlive(KeepAlive(timeSeconds = 30))
+            .build()
 
         mqttClient.connect(connectOptions)
     }
 
     private fun initialiseCourier() {
         val mqttConfig = MqttV3Configuration(
-            socketFactory = null,
             logger = getLogger(),
-            eventHandler = eventHandler,
             authenticator = object : Authenticator {
                 override fun authenticate(
                     connectOptions: MqttConnectOptions,
                     forceRefresh: Boolean
                 ): MqttConnectOptions {
-                    return connectOptions.copy(password = password.text.toString())
+                    return connectOptions.newBuilder()
+                        .password(password.text.toString())
+                        .build()
                 }
             },
             mqttInterceptorList = listOf(MqttChuckInterceptor(this, MqttChuckConfig(retentionPeriod = Period.ONE_HOUR))),
@@ -133,10 +131,12 @@ class MainActivity : AppCompatActivity() {
                 activityCheckIntervalSeconds = 30,
                 incomingMessagesTTLSecs = 60,
                 incomingMessagesCleanupIntervalSecs = 10,
+                maxInflightMessagesLimit = 1000,
             ),
             pingSender = WorkPingSenderFactory.createMqttPingSender(applicationContext, WorkManagerPingSenderConfig())
         )
         mqttClient = MqttClientFactory.create(this, mqttConfig)
+        mqttClient.addEventHandler(eventHandler)
 
         val configuration = Courier.Configuration(
             client = mqttClient,

build.gradle.kts

@@ -56,6 +56,7 @@ val clean by tasks.creating(Delete::class) {
     delete("${rootDir}/courier/build")
     delete("${rootDir}/courier-core/build")
     delete("${rootDir}/courier-core-android/build")
+    delete("${rootDir}/courier-message-adapter-text/build")
     delete("${rootDir}/courier-message-adapter-gson/build")
     delete("${rootDir}/courier-message-adapter-moshi/build")
     delete("${rootDir}/courier-message-adapter-protobuf/build")

buildSrc/src/main/kotlin/deps.kt

@@ -59,6 +59,7 @@ object deps {
             const val runner = "androidx.test:runner:1.2.0"
             const val roboelectric = "org.robolectric:robolectric:4.2"
             const val mockito = "com.nhaarman.mockitokotlin2:mockito-kotlin:2.2.0"
+            const val mockitoCore = "org.mockito:mockito-core:4.4.0"
             const val junitExt = "androidx.test.ext:junit:1.1.1"
             const val kotlinTestJunit = "org.jetbrains.kotlin:kotlin-test-junit:${versions.kotlin}"
         }

chuck-mqtt/src/main/java/com/gojek/chuckmqtt/internal/presentation/mapper/MqttTransactionUiModelMapper.kt

@@ -5,8 +5,6 @@ import com.gojek.chuckmqtt.internal.domain.model.MqttTransactionDomainModel
 import com.gojek.chuckmqtt.internal.presentation.model.MqttTransactionUiModel
 import com.gojek.chuckmqtt.internal.utils.formatBody
 import com.gojek.chuckmqtt.internal.utils.formatByteCount
-import java.text.DateFormat
-import kotlin.text.Charsets.UTF_8
 import `in`.mohalla.paho.client.mqttv3.MqttMessage
 import `in`.mohalla.paho.client.mqttv3.internal.wire.MqttConnack
 import `in`.mohalla.paho.client.mqttv3.internal.wire.MqttConnect
@@ -23,8 +21,11 @@ import `in`.mohalla.paho.client.mqttv3.internal.wire.MqttSubscribe
 import `in`.mohalla.paho.client.mqttv3.internal.wire.MqttUnsubAck
 import `in`.mohalla.paho.client.mqttv3.internal.wire.MqttUnsubscribe
 import `in`.mohalla.paho.client.mqttv3.internal.wire.MqttWireMessage
+import java.text.DateFormat
+import kotlin.text.Charsets.UTF_8
 
-internal class MqttTransactionUiModelMapper : Mapper<MqttTransactionDomainModel, MqttTransactionUiModel> {
+internal class MqttTransactionUiModelMapper :
+    Mapper<MqttTransactionDomainModel, MqttTransactionUiModel> {
     override fun map(input: MqttTransactionDomainModel): MqttTransactionUiModel {
         return with(input) {
             MqttTransactionUiModel(
@@ -161,7 +162,8 @@ internal class MqttTransactionUiModelMapper : Mapper<MqttTransactionDomainModel,
         return when (mqttWireMessage) {
             is MqttPublish -> {
                 formatBody(String(mqttWireMessage.message.payload, UTF_8))
-            } else -> ""
+            }
+            else -> ""
         }
     }
 
@@ -202,7 +204,7 @@ internal class MqttTransactionUiModelMapper : Mapper<MqttTransactionDomainModel,
             is MqttPublish -> {
                 sb.append("PUBLISH \n")
                 sb.append("Qos : ${mqttWireMessage.message.qos} \n")
-                if (mqttWireMessage.message.qos > 0) {
+                if (mqttWireMessage.message.qos > 0 || mqttWireMessage.message.type > 2) {
                     sb.append("MsgId : ${mqttWireMessage.messageId} \n")
                 }
                 sb.append("Retained : ${mqttWireMessage.message.isRetained} \n")

courier-auth-http/build.gradle.kts

@@ -30,6 +30,7 @@ dependencies {
 
     implementation(deps.square.retrofit)
 
+    testImplementation(deps.android.test.mockitoCore)
     testImplementation(deps.android.test.kotlinTestJunit)
 }
 

courier-core/api/courier-core.api

@@ -8,8 +8,9 @@ public final class com/gojek/courier/Message$Bytes : com/gojek/courier/Message {
 }
 
 public abstract interface class com/gojek/courier/MessageAdapter {
-	public abstract fun fromMessage (Lcom/gojek/courier/Message;)Ljava/lang/Object;
-	public abstract fun toMessage (Ljava/lang/Object;)Lcom/gojek/courier/Message;
+	public abstract fun contentType ()Ljava/lang/String;
+	public abstract fun fromMessage (Ljava/lang/String;Lcom/gojek/courier/Message;)Ljava/lang/Object;
+	public abstract fun toMessage (Ljava/lang/String;Ljava/lang/Object;)Lcom/gojek/courier/Message;
 }
 
 public abstract interface class com/gojek/courier/MessageAdapter$Factory {
@@ -18,8 +19,11 @@ public abstract interface class com/gojek/courier/MessageAdapter$Factory {
 
 public final class com/gojek/courier/QoS : java/lang/Enum {
 	public static final field ONE Lcom/gojek/courier/QoS;
+	public static final field ONE_WITHOUT_PERSISTENCE_AND_NO_RETRY Lcom/gojek/courier/QoS;
+	public static final field ONE_WITHOUT_PERSISTENCE_AND_RETRY Lcom/gojek/courier/QoS;
 	public static final field TWO Lcom/gojek/courier/QoS;
 	public static final field ZERO Lcom/gojek/courier/QoS;
+	public final fun getType ()I
 	public final fun getValue ()I
 	public static fun valueOf (Ljava/lang/String;)Lcom/gojek/courier/QoS;
 	public static fun values ()[Lcom/gojek/courier/QoS;

courier-core/src/main/java/com/gojek/courier/MessageAdapter.kt

@@ -5,10 +5,13 @@ import java.lang.reflect.Type
 interface MessageAdapter<T> {
 
     /** Returns an object of type `T` that represents a [Message]. */
-    fun fromMessage(message: Message): T
+    fun fromMessage(topic: String, message: Message): T
 
     /** Returns a [Message] that represents [data]. */
-    fun toMessage(data: T): Message
+    fun toMessage(topic: String, data: T): Message
+
+    /** Returns the content type supported by this adapter. */
+    fun contentType(): String
 
     /** Creates [MessageAdapter] instances based on a type and target usage. */
     interface Factory {

courier-core/src/main/java/com/gojek/courier/QoS.kt

@@ -1,7 +1,21 @@
 package com.gojek.courier
 
-enum class QoS(val value: Int) {
-    ZERO(0),
-    ONE(1),
-    TWO(2)
+enum class QoS(val value: Int, val type: Int) {
+    // http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html#_Toc398718100
+    ZERO(0, 0),
+
+    // http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html#_Toc398718100
+    ONE(1, 1),
+
+    // http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html#_Toc398718100
+    TWO(2, 2),
+
+    /** Like QoS1, Message delivery is acknowledged with PUBACK, but unlike QoS1 messages are
+     neither persisted nor retried at send after one attempt.
+     The message arrives at the receiver either once or not at all **/
+    ONE_WITHOUT_PERSISTENCE_AND_NO_RETRY(0, 3),
+
+    /** Like QoS1, Message delivery is acknowledged with PUBACK, but unlike QoS1 messages are
+     not persisted. The messages are retried within active connection if delivery is not acknowledged.**/
+    ONE_WITHOUT_PERSISTENCE_AND_RETRY(0, 4)
 }

courier-message-adapter-gson/src/main/java/com/gojek/courier/messageadapter/gson/GsonMessageAdapter.kt

@@ -19,15 +19,15 @@ private class GsonMessageAdapter<T> constructor(
     private val typeAdapter: TypeAdapter<T>
 ) : MessageAdapter<T> {
 
-    override fun fromMessage(message: Message): T {
+    override fun fromMessage(topic: String, message: Message): T {
         val stringValue = when (message) {
             is Message.Bytes -> String(message.value)
         }
         val jsonReader = gson.newJsonReader(StringReader(stringValue))
         return typeAdapter.read(jsonReader)!!
     }
 
-    override fun toMessage(data: T): Message {
+    override fun toMessage(topic: String, data: T): Message {
         val buffer = Buffer()
         val writer = OutputStreamWriter(buffer.outputStream(), UTF_8)
         val jsonWriter = gson.newJsonWriter(writer)
@@ -36,6 +36,8 @@ private class GsonMessageAdapter<T> constructor(
         val stringValue = buffer.readByteString().utf8()
         return Message.Bytes(stringValue.toByteArray())
     }
+
+    override fun contentType() = "application/json"
 }
 
 class GsonMessageAdapterFactory(

courier-message-adapter-moshi/src/main/java/com/gojek/courier/messageadapter/moshi/MoshiMessageAdapter.kt

@@ -16,7 +16,7 @@ private class MoshiMessageAdapter<T> constructor(
     private val jsonAdapter: JsonAdapter<T>
 ) : MessageAdapter<T> {
 
-    override fun fromMessage(message: Message): T {
+    override fun fromMessage(topic: String, message: Message): T {
         val stringValue = when (message) {
             is Message.Bytes -> {
                 val byteString = ByteString.of(message.value, 0, message.value.size)
@@ -32,11 +32,13 @@ private class MoshiMessageAdapter<T> constructor(
         return jsonAdapter.fromJson(stringValue)!!
     }
 
-    override fun toMessage(data: T): Message {
+    override fun toMessage(topic: String, data: T): Message {
         val stringValue = jsonAdapter.toJson(data)
         return Message.Bytes(stringValue.toByteArray())
     }
 
+    override fun contentType() = "application/json"
+
     private companion object {
         private val UTF8_BOM = ByteString.decodeHex("EFBBBF")
     }

courier-message-adapter-protobuf/src/main/java/com/gojek/courier/messageadapter/protobuf/ProtobufMessageAdapter.kt

@@ -17,7 +17,7 @@ private class ProtobufMessageAdapter<T : MessageLite> constructor(
     private val registry: ExtensionRegistryLite?
 ) : MessageAdapter<T> {
 
-    override fun fromMessage(message: Message): T {
+    override fun fromMessage(topic: String, message: Message): T {
         val bytesValue = when (message) {
             is Message.Bytes -> message.value
         }
@@ -31,7 +31,9 @@ private class ProtobufMessageAdapter<T : MessageLite> constructor(
         }
     }
 
-    override fun toMessage(data: T): Message = Message.Bytes(data.toByteArray())
+    override fun toMessage(topic: String, data: T): Message = Message.Bytes(data.toByteArray())
+
+    override fun contentType() = "application/x-protobuf"
 }
 
 class ProtobufMessageAdapterFactory(

courier-message-adapter-text/.gitignore

@@ -0,0 +1 @@
+/build

courier-message-adapter-text/api/courier-message-adapter-text.api

@@ -0,0 +1,5 @@
+public final class com/gojek/courier/messageadapter/text/TextMessageAdapterFactory : com/gojek/courier/MessageAdapter$Factory {
+	public fun <init> ()V
+	public fun create (Ljava/lang/reflect/Type;[Ljava/lang/annotation/Annotation;)Lcom/gojek/courier/MessageAdapter;
+}
+