Hello Gojek Courier team,

I hope this message finds you well. I'm reaching out to suggest a feature enhancement that could streamline the handling of multi-topic subscriptions within the service interface of Gojek Courier.

Current Situation:
Currently, in order to subscribe to multiple topics using the @SubscribeMultiple annotation, we are required to create separate @Receive annotations for each topic. While this approach works, it can lead to redundant code and reduced maintainability, especially when the response models for these topics are identical.

Proposal:
I'd like to propose the introduction of a "Global Listener" concept that would allow us to handle multiple topic subscriptions with consistent response models in a more concise and elegant manner.

Suggested Implementation:

1. Introduction of Global Listener Annotation: Add a new annotation, say @GlobalListener, that can be placed at the service interface level to denote a method that will act as a global listener.

2. Reduced Redundancy with Shared Response Model: With this new annotation, the service interface method could be designed to listen to multiple topics using wildcard expressions or dynamic parameters, eliminating the need to define separate @Receive annotations for each topic with the same response model.

3. Example:

   @Service
   interface CourierService {
       @GlobalListener
       @Receive
       fun handleGlobalTopic(@TopicMap topics: List<String>) : Model
   }

Benefits:

• Code Elegance: The service interface becomes more concise and easier to manage, as it can handle multiple topic subscriptions with a single method.
• Consistency: When topics share the same response model, there's no need to duplicate the @Receive annotations and response model declarations.
• Maintenance: Changes to the response model or subscription logic can be applied globally, minimizing the chances of errors due to manual code replication.

Additional Consideration:
The proposed "Global Listener" should be flexible enough to handle dynamic parameters or wildcards in topic patterns, ensuring it can cater to a wide range of use cases.

Conclusion:
I believe that implementing a "Global Listener" mechanism as described above could significantly enhance the efficiency and maintainability of the Gojek Courier service interface. It would streamline the process of subscribing to multiple topics with consistent response models, while also adhering to best practices for reducing redundancy in code.

Thank you for considering this feature request. I'm looking forward to your feedback and thoughts on this idea.

Best Regards,
M E Mohit

There is a crash when we try to connect to an invalid topic.
For example home# or home+

Steps to reproduce -:

1. Connect to a MQTT broker
2. Subscribe to an invalid topic name. Example home#

Error Logs -:

2022-07-16 13:45:37.778 10608-10647/com.gojek.courier.app E/AndroidRuntime: FATAL EXCEPTION: MQTT_Thread
    Process: com.gojek.courier.app, PID: 10608
    java.lang.IllegalArgumentException: Invalid usage of multi-level wildcard in topic string: home#
        at org.eclipse.paho.client.mqttv3.MqttTopic.validate(MqttTopic.java:200)
        at org.eclipse.paho.client.mqttv3.MqttAsyncClient.subscribe(MqttAsyncClient.java:889)
        at com.gojek.mqtt.connection.MqttConnection.subscribe(MqttConnection.kt:474)
        at com.gojek.mqtt.client.v3.impl.AndroidMqttClient.subscribeMqtt(AndroidMqttClient.kt:465)
        at com.gojek.mqtt.scheduler.runnable.SubscribeRunnable.run(SubscribeRunnable.kt:11)
        at android.os.Handler.handleCallback(Handler.java:938)
        at android.os.Handler.dispatchMessage(Handler.java:99)
        at android.os.Looper.loopOnce(Looper.java:201)
        at android.os.Looper.loop(Looper.java:288)
        at android.os.HandlerThread.run(HandlerThread.java:67)

Video for your reference

If we want to connect to courier for a short duration and then dispose it, since the HandlerThread is never stopped it is always active till the app is killed.

As we use primitive type on the data class properties declaration it would give a high chance for error-prone of doing substitution in between properties. Indeed naturally we're able to avoid this issue by declaring the name of the properties themselves when assigning the value. However, it still has a chance the users do not follow the convention for doing that. The consideration for using the inline/value class can be helpful here.

Hello! Thanks for open-sourcing this library. The features added on top of Paho look great!

Are there any plans to support MQTT v5?

As we have seen in the example, the initialization of Courier took few configurations. Since most of them are a default configuration. would be better if we create a bean in the configuration class itself, and the caller is able to mutate as per need. doing this would less overhead for setup the configuration

Example:

data class FooConfiguration(
    val arg1: String,
    val arg2: String,
    val arg3: String
) {

    fun default(): FooConfiguration {
        return FooConfiguration(/*args*/, /*args*/, /*args*/)
    }
}

// client side
val foo = FooConfiguration.default().copy(arg1 = "magic")

I am trying to connect to AWS IoT MQTT broker using courier library with custom authentication. It gets connected via WSS scheme but not with SSL or TLS scheme.

Caused by android.net.ConnectivityManager$TooManyRequestsException
android.net.ConnectivityManager.convertServiceException (ConnectivityManager.java:3609)

android.net.ConnectivityManager.registerDefaultNetworkCallback (ConnectivityManager.java:4256)
com.gojek.networktracker.NetworkStateTrackerImpl.startTracking$network_tracker_release (NetworkStateTrackerImpl.java:115)
com.gojek.networktracker.NetworkStateTrackerImpl.addListener (NetworkStateTrackerImpl.java:86)
com.gojek.keepalive.OptimalKeepAliveCalculator. (OptimalKeepAliveCalculator.java:31)
com.gojek.keepalive.KeepAliveCalculatorFactory.create (KeepAliveCalculatorFactory.java:29)
com.gojek.keepalive.OptimalKeepAliveProvider. (OptimalKeepAliveProvider.java:16)
com.gojek.keepalive.OptimalKeepAliveProvider. (OptimalKeepAliveProvider.java:9)
com.gojek.mqtt.client.internal.MqttClientInternal.initialiseOptimalKeepAliveProvider (MqttClientInternal.java:157)
com.gojek.mqtt.client.internal.MqttClientInternal.initialiseAdaptiveMqttClient (MqttClientInternal.java:122)
com.gojek.mqtt.client.internal.MqttClientInternal. (MqttClientInternal.java:64)

Thanks for this much needed android library now that paho project is abandoned.

With Android 13, setExactAndAllowWhileIdle() permission is now protected permission & only alarm apps can use it.

Can the connection with broker reliably survive in background without this permission? What are your plans to target Android 13 compatibility.

TIA

When subscribing for a topic if at the broker end there is a timeout the broker gives a SubAck with granted Qos as 128. This event is never propagated to the consumer by the courier lib.

@deepanshu42

The connection retry's continuously for every 1 second indefinitely when there is an authentication failure.

Even after the retry policies set and and also using default retry policy doesn't not exponentially retry the connection.

After setting the authFailureHandler = AuthenticatorHandler() when creating MqttV3Configuration, this will help not retry more than once.

But want to know if that's an expected behaviour to retry every 1 second indefinitely.

@deepanshu42 I was observing a crash in Coordinator due to the following code in the SDK, if you see the block topics.size == 1 it would unsubscribe to the first topic, which is fine, but let's say, the topic size is zero, then the else querying the topics[0] will throw ArrayIndexOutOfBoundsException,

This is the same even for subscribeAll too, please check

override fun unsubscribe(stubMethod: StubMethod.Unsubscribe, args: Array<Any>): Any {
        logger.d("Coordinator", "Unsubscribe method invoked")
        stubMethod.argumentProcessor.inject(args)
        val topics = stubMethod.argumentProcessor.getTopics()
        val status = if (topics.size == 1) {
            client.unsubscribe(topics[0])
        } else {
            client.unsubscribe(topics[0], *topics.sliceArray(IntRange(1, topics.size - 1)))
        }
        logger.d("Coordinator", "Unsubscribing topics: $topics")
        return status
    }

Crash stacktrace

Fatal Exception: java.lang.ArrayIndexOutOfBoundsException: length=0; index=0
       at com.gojek.courier.coordinator.Coordinator.unsubscribe(Coordinator.kt:133)
       at com.gojek.courier.stub.StubInterface.invoke(StubInterface.kt:32)
       at com.gojek.courier.stub.ProxyFactory.createInvocationHandler$lambda-0(ProxyFactory.kt:41)
       at java.lang.reflect.Proxy.invoke(Proxy.java:1006)
       at $Proxy48.unsubscribe()
       at okyi.okyi.okyi.XNkf.kDyL.OEVx.OEVx.HVAU$IwUN.invokeSuspend(OperatorImpl.kt:21)
       at okyi.okyi.okyi.XNkf.kDyL.OEVx.OEVx.HVAU$IwUN.invoke(OperatorImpl.kt:2)
       at kotlinx.coroutines.flow.ChannelFlowBuilder.collectTo$suspendImpl(Builders.kt:320)
       at kotlinx.coroutines.flow.ChannelFlowBuilder.collectTo(Builders.kt)
       at kotlinx.coroutines.flow.CallbackFlowBuilder.collectTo(Builders.kt:334)
       at kotlinx.coroutines.flow.internal.ChannelFlow$collectToFun$1.invokeSuspend(ChannelFlow.kt:60)
       at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:33)
       at kotlinx.coroutines.DispatchedTask.run(DispatchedTask.kt:108)
       at kotlinx.coroutines.internal.LimitedDispatcher$Worker.run(LimitedDispatcher.kt:115)
       at kotlinx.coroutines.scheduling.TaskImpl.run(Tasks.kt:103)
       at kotlinx.coroutines.scheduling.CoroutineScheduler.runSafely(CoroutineScheduler.kt:584)
       at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.executeTask(CoroutineScheduler.kt:793)
       at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.runWorker(CoroutineScheduler.kt:697)
       at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.run(CoroutineScheduler.kt)

How to get a message id for a published message to MQTT.

Use case -> i have to perform some operations based on the message id in my DB to ensure there's no data loss.

@Receive(topic = “{topic}“)
fun receive(@Path(“topic”) topic: String): Observable<String>

when the topic is "some/word/+/foo" , it can not receive message from "some/word/1234/foo" . i think this happen because courier try to listen topic "some/word/+/foo" with actual '+', not wildcard.

but, from interceptor, courier successfully receive the message from the topic.