Allow configuration of maximum number of in-flight messages fo the underlying paho MQTT client

[mduesterhoeft]

Which AWS Services is the feature request for?

IoT

Is your feature request related to a problem? Please describe.

We see the following exception quite frequently:

com.amazonaws.AmazonClientException: Client error while publishing.
       at com.amazonaws.mobileconnectors.iot.AWSIotMqttManager.publishData(AWSIotMqttManager.java:1407)
       at io.moia.iot.IotConnection$publishData$2.invokeSuspend(IotConnection.kt:955)
       at io.moia.iot.IotConnection$publishData$2.invoke(:10)
       at io.moia.iot.IotConnection.retry(IotConnection.kt:941)
       at io.moia.iot.IotConnection$retry$1.invokeSuspend(:18)
       at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:33)
       at kotlinx.coroutines.DispatchedTask.run(Dispatched.kt:241)
       at kotlinx.coroutines.scheduling.CoroutineScheduler.runSafely(CoroutineScheduler.kt:594)
       at kotlinx.coroutines.scheduling.CoroutineScheduler.access$runSafely(CoroutineScheduler.kt:60)
       at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.run(CoroutineScheduler.kt:740)
Caused by org.eclipse.paho.client.mqttv3.MqttException: Too many publishes in progress
       at org.eclipse.paho.client.mqttv3.internal.ClientState.send(ClientState.java:524)
       at org.eclipse.paho.client.mqttv3.internal.ClientComms.internalSend(ClientComms.java:161)
       at org.eclipse.paho.client.mqttv3.internal.ClientComms.sendNoWait(ClientComms.java:191)
       at org.eclipse.paho.client.mqttv3.MqttAsyncClient.publish(MqttAsyncClient.java:1251)
       at org.eclipse.paho.client.mqttv3.MqttAsyncClient.publish(MqttAsyncClient.java:1200)
       at com.amazonaws.mobileconnectors.iot.AWSIotMqttManager.publishData(AWSIotMqttManager.java:1405)
       at io.moia.iot.IotConnection$publishData$2.invokeSuspend(IotConnection.kt:955)
       at io.moia.iot.IotConnection$publishData$2.invoke(:10)
       at io.moia.iot.IotConnection.retry(IotConnection.kt:941)
       at io.moia.iot.IotConnection$retry$1.invokeSuspend(:18)
       at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:33)
       at kotlinx.coroutines.DispatchedTask.run(Dispatched.kt:241)
       at kotlinx.coroutines.scheduling.CoroutineScheduler.runSafely(CoroutineScheduler.kt:594)
       at kotlinx.coroutines.scheduling.CoroutineScheduler.access$runSafely(CoroutineScheduler.kt:60)
       at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.run(CoroutineScheduler.kt:740)

This is the JavaDoc I could find in the Paho MQTT client for this exception (see MqttClientException.)

/**
 * A request has been made to send a message but the maximum number of inflight 
 * messages has already been reached. Once one or more messages have been moved
 * then new messages can be sent.   
 */
public static final short REASON_CODE_MAX_INFLIGHT              = 32202;

The default for the maximum number of in-flight messages is 10. The underlying Paho MQTT client allows us to customize this value. But the AWSMqttManager wraps this and does not allow to configure it from the outside.

Describe the solution you'd like Expose the configuration option regarding maximum number if in-flight messages that PAHO MQTT already offers (see MqttConnectOptions.)

[jamesonwilliams]

@mduesterhoeft Thanks for the idea, sounds reasonable. We have a number of things in flight and will track demand on this. Others: please add a 👍 if you'd like to see this implemented.

[MrLiuYunPing]

Where is the corresponding IOS set

[eduprat-chwy]

Any news about this? I'm having the same crash, @mduesterhoeft could you fix it in some way?

[mayank1jain]

Any update on this ? I have same crash.

Is there any way to retry publish later if it fails to publish ?

[eeatonaws]

No new updates on this feature request at this time. A workaround is to add a delay between publishing messages to give additional time for in-flight messages to be delivered or use the version of AWSIotMqttManager.publishString(...) with a message callback that indicates whether the message was delivered successfully or failed and needs to be resent.