Spring Integration hits a bug in Paho client with $share/ MQTT subscriptions

[vhbcm]

Spring Integration hits a bug in Paho client with $share/ MQTT subscriptions

In what version(s) of Spring Integration are you seeing this issue?

For example:

6.1.5

Describe the bug

Spring Integration hits a bug (eclipse-paho/paho.mqtt.java/issues/827) in Paho MQTT client with $share/ MQTT subscriptions. This bug can be worked around by using subscription ID by providing it in MqttProperties when calling, e.g., org.eclipse.paho.mqttv5.client.IMqttAsyncClient#subscribe(org.eclipse.paho.mqttv5.common.MqttSubscription[], java.lang.Object, org.eclipse.paho.mqttv5.client.MqttActionListener, org.eclipse.paho.mqttv5.client.IMqttMessageListener, org.eclipse.paho.mqttv5.common.packet.MqttProperties). Unfortunately, Spring Integration currently calls only org.eclipse.paho.mqttv5.client.IMqttAsyncClient#subscribe(org.eclipse.paho.mqttv5.common.MqttSubscription, org.eclipse.paho.mqttv5.client.IMqttMessageListener) from org.springframework.integration.mqtt.inbound.Mqttv5PahoMessageDrivenChannelAdapter#addTopic and does not provide subscription ID.

To Reproduce

Subscribe on $share/sharegroup/topic using Mqttv5PahoMessageDrivenChannelAdapter and send message to topic. The message will be received but it will be rejected by Paho itself because it is comparing the filter and the message topic textually without removing the $share/sharegroup/ prefix.

Expected behavior

Mqttv5PahoMessageDrivenChannelAdapter should be able to subscribe to shared MQTT topics.

Sample

N/A

[artembilan]

Are you referring to this property:

	/**
	 * Subscription Identifier. (Subscribe Only) See
	 * {@link MqttProperties#getSubscriptionIdentifier()}
	 * 
	 * @param subscriptionIdentifier
	 *            The Subscription Identifier.
	 */
	public void setSubscriptionIdentifier(Integer subscriptionIdentifier) {

?

I think there might be some other props which could be set for this or that use-case.
So, I guess something like Mqttv5PahoMessageDrivenChannelAdapter.setSubscriptionMqttProperties(MqttProperties) would be sufficient.
Although I see on this sub-set of valid props:

	private static final Byte[] validProperties = { MqttProperties.SUBSCRIPTION_IDENTIFIER,
			MqttProperties.SUBSCRIPTION_IDENTIFIER_SINGLE,
			MqttProperties.USER_DEFINED_PAIR_IDENTIFIER };

So, I'm not sure if we have to expose the whole MqttProperties or just subscriptionIdentifier on the Mqttv5PahoMessageDrivenChannelAdapter.
Although with the former we would nail any future potential requests for those properties.

If that would work as a workaround, then I'm more than happy to accept such a contribution.

Thanks

[wilx]

Are you referring to this property:

	/**
	 * Subscription Identifier. (Subscribe Only) See
	 * {@link MqttProperties#getSubscriptionIdentifier()}
	 * 
	 * @param subscriptionIdentifier
	 *            The Subscription Identifier.
	 */
	public void setSubscriptionIdentifier(Integer subscriptionIdentifier) {

?

Yes.

I think there might be some other props which could be set for this or that use-case. So, I guess something like Mqttv5PahoMessageDrivenChannelAdapter.setSubscriptionMqttProperties(MqttProperties) would be sufficient. Although I see on this sub-set of valid props:

	private static final Byte[] validProperties = { MqttProperties.SUBSCRIPTION_IDENTIFIER,
			MqttProperties.SUBSCRIPTION_IDENTIFIER_SINGLE,
			MqttProperties.USER_DEFINED_PAIR_IDENTIFIER };

So, I'm not sure if we have to expose the whole MqttProperties or just subscriptionIdentifier on the Mqttv5PahoMessageDrivenChannelAdapter. Although with the former we would nail any future potential requests for those properties.

I am not sure if it needs to be exposed at all. The subscription identifier can be an internal detail. It is only useful for the client's internal mapping, I think. At least in my use case.

If that would work as a workaround, then I'm more than happy to accept such a contribution.

Thanks

[artembilan]

Well, that would be easier, but then we need to understand what is that number supposed to be.
Some hash from the getClientId()? Some random value we chose on start?
Well, we still may have it to be set externally if anyway has some use-case where they need to be able to identify those subscription by some other meaning.
And have it defaulted to something random or that hash.

The point is that we might need to back-port this to previous version to provide the support for lost functionality.

Here is some discussion as well: https://stackoverflow.com/questions/75950944/shared-subscription-with-spring-integration-mqttv5-not-receiving-messages

[wilx]

The subscription identifier is an integer:

3.8.2.1.2 Subscription Identifier

11 (0x0B) Byte, Identifier of the Subscription Identifier.

Followed by a Variable Byte Integer representing the identifier of the subscription. The Subscription Identifier can have the value of 1 to 268,435,455. It is a Protocol Error if the Subscription Identifier has a value of 0. It is a Protocol Error to include the Subscription Identifier more than once.