Updates from gojek/courier-android

.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,30 @@ 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))),

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}"
         }

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 {

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-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;
+}
+

courier-message-adapter-text/build.gradle.kts

@@ -0,0 +1,28 @@
+import plugin.KotlinLibraryConfigurationPlugin
+
+apply<KotlinLibraryConfigurationPlugin>()
+apply("$rootDir/gradle/script-ext.gradle")
+
+val version = ext.get("gitVersionName")
+
+ext {
+    set("PUBLISH_GROUP_ID", "com.gojek.courier")
+    set("PUBLISH_ARTIFACT_ID", "courier-message-adapter-text")
+    set("PUBLISH_VERSION", ext.get("gitVersionName"))
+    set("minimumCoverage", "0.0")
+}
+
+plugins {
+    id("java-library")
+    kotlin("jvm")
+    id(ScriptPlugins.apiValidator) version versions.apiValidator
+}
+
+dependencies {
+    api(project(":courier-core"))
+    implementation(deps.kotlin.stdlib.core)
+    implementation(deps.square.okio)
+    testImplementation(deps.android.test.kotlinTestJunit)
+}
+
+apply(from = "${rootProject.projectDir}/gradle/publish-module.gradle")

courier-message-adapter-text/src/main/java/com/gojek/courier/messageadapter/text/TextMessageAdapter.kt

@@ -0,0 +1,23 @@
+package com.gojek.courier.messageadapter.text
+
+import com.gojek.courier.Message
+import com.gojek.courier.MessageAdapter
+import com.gojek.courier.utils.getRawType
+import java.lang.reflect.Type
+
+class TextMessageAdapterFactory : MessageAdapter.Factory {
+    override fun create(type: Type, annotations: Array<Annotation>): MessageAdapter<*> = when (type.getRawType()) {
+        String::class.java -> TextMessageAdapter()
+        else -> throw IllegalArgumentException("Type is not supported by this MessageAdapterFactory: $type")
+    }
+}
+
+internal class TextMessageAdapter : MessageAdapter<String> {
+    override fun fromMessage(topic: String, message: Message): String = when (message) {
+        is Message.Bytes -> String(message.value)
+    }
+
+    override fun toMessage(topic: String, data: String): Message = Message.Bytes(data.toByteArray())
+
+    override fun contentType() = "text/plain"
+}

courier-message-adapter-text/src/test/java/com/gojek/courier/messageadapter/text/ExampleUnitTest.kt

@@ -0,0 +1,16 @@
+package com.gojek.courier.messageadapter.gson
+
+import org.junit.Assert.assertEquals
+import org.junit.Test
+
+/**
+ * Example local unit test, which will execute on the development machine (host).
+ *
+ * See [testing documentation](http://d.android.com/tools/testing).
+ */
+class ExampleUnitTest {
+    @Test
+    fun addition_isCorrect() {
+        assertEquals(4, 2 + 2)
+    }
+}

courier/src/main/java/com/gojek/courier/Courier.kt

@@ -3,8 +3,6 @@ package com.gojek.courier
 import com.gojek.courier.coordinator.Coordinator
 import com.gojek.courier.logging.ILogger
 import com.gojek.courier.logging.NoOpLogger
-import com.gojek.courier.messageadapter.builtin.BuiltInMessageAdapterFactory
-import com.gojek.courier.streamadapter.builtin.BuiltInStreamAdapterFactory
 import com.gojek.courier.stub.ProxyFactory
 import com.gojek.courier.stub.StubInterface
 import com.gojek.courier.stub.StubMethod
@@ -52,10 +50,10 @@ class Courier(configuration: Configuration) {
     )
 
     private fun Configuration.createStreamAdapterResolver(): StreamAdapterResolver {
-        return StreamAdapterResolver(listOf(BuiltInStreamAdapterFactory()) + streamAdapterFactories)
+        return StreamAdapterResolver(streamAdapterFactories)
     }
 
     private fun Configuration.createMessageAdapterResolver(): MessageAdapterResolver {
-        return MessageAdapterResolver(listOf(BuiltInMessageAdapterFactory()) + messageAdapterFactories)
+        return MessageAdapterResolver(messageAdapterFactories)
     }
 }