GH-10760: Support topic filters in messages arrived callback (#10761)

* GH-10760: Support topic filters in messages arrived callback

Fixes: #10760

The previous fix with protection against subscribed topics did not include
the logic about wildcard support in MQTT subscriptions

* Fix `MqttPahoMessageDrivenChannelAdapter` extracting `messageArrivedIfMatched()`
which would be called as a fallback common callback.
Delegate matching logic to the `MqttTopic.isMatched()` utility
* Fix `Mqttv5PahoMessageDrivenChannelAdapter` extracting `messageArrivedIfMatched()`
which would be called as a fallback common callback.
Delegate matching logic to the `MqttTopicValidator.isMatched()` utility.
However, before that check for the `$share/` prefix and strip two subscription parts.
Exactly the logic missed in Paho Client
* Modify `Mqttv5BackToBackTests.testSharedTopicMqttv5Interaction()` for wildcard to be sure
new matching logic works as expected
* Some MQTT MDChAs cleanup after review

Auto-cherry-pick to `7.0.x` & `6.5.x`**

Author: artembilan

spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/MqttPahoMessageDrivenChannelAdapter.java

@@ -29,6 +29,7 @@
 import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
 import org.eclipse.paho.client.mqttv3.MqttException;
 import org.eclipse.paho.client.mqttv3.MqttMessage;
+import org.eclipse.paho.client.mqttv3.MqttTopic;
 
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.integration.IntegrationMessageHeaderAccessor;
@@ -72,7 +73,7 @@ public class MqttPahoMessageDrivenChannelAdapter
 	 * Never invoked because MQTT v3 does not support shared subscriptions.
 	 * But it is here for consistency with the {@link ClientManager} requirements
 	 */
-	private final ClientManager.DefaultMessageHandler<MqttMessage> defaultMessageHandler = this::messageArrived;
+	private final ClientManager.DefaultMessageHandler<MqttMessage> defaultMessageHandler = this::messageArrivedIfMatched;
 
 	private final MqttPahoClientFactory clientFactory;
 
@@ -188,7 +189,6 @@ protected void doStart() {
 		}
 	}
 
-	@SuppressWarnings("deprecation")
 	private void connect() throws MqttException {
 		this.lock.lock();
 		try {
@@ -374,14 +374,6 @@ public void connectionLost(Throwable cause) {
 
 	@Override
 	public void messageArrived(String topic, MqttMessage mqttMessage) {
-		// Just simple check since MQTT v3 does not support shared subscriptions.
-		boolean subscribed = Arrays.asList(getTopic()).contains(topic);
-		if (!subscribed) {
-			logger.trace(() ->
-					"Arrived message on topic '" + topic + "' this channel adapter is not subscribed to. Ignoring...");
-			return;
-		}
-
 		AbstractIntegrationMessageBuilder<?> builder = toMessageBuilder(topic, mqttMessage);
 		if (builder != null) {
 			if (isManualAcks()) {
@@ -451,6 +443,17 @@ public void connectComplete(boolean reconnect, String serverURI) {
 		}
 	}
 
+	private void messageArrivedIfMatched(String topic, MqttMessage mqttMessage) {
+		for (String subscribedTopic : getTopic()) {
+			if (MqttTopic.isMatched(subscribedTopic, topic)) {
+				messageArrived(topic, mqttMessage);
+				return;
+			}
+		}
+		logger.trace(() ->
+				"Arrived message on topic '" + topic + "' this channel adapter is not subscribed to. Ignoring...");
+	}
+
 	/**
 	 * Used to complete message arrival when {@link #isManualAcks()} is true.
 	 *

spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java

@@ -37,6 +37,7 @@
 import org.eclipse.paho.mqttv5.common.MqttMessage;
 import org.eclipse.paho.mqttv5.common.MqttSubscription;
 import org.eclipse.paho.mqttv5.common.packet.MqttProperties;
+import org.eclipse.paho.mqttv5.common.util.MqttTopicValidator;
 
 import org.springframework.beans.factory.BeanCreationException;
 import org.springframework.context.ApplicationEventPublisher;
@@ -94,7 +95,7 @@ public class Mqttv5PahoMessageDrivenChannelAdapter
 	/**
 	 * Used as a fallback option for shared subscriptions unrouted messages
 	 */
-	private final ClientManager.DefaultMessageHandler<MqttMessage> defaultMessageHandler = this::messageArrived;
+	private final ClientManager.DefaultMessageHandler<MqttMessage> defaultMessageHandler = this::messageArrivedIfMatched;
 
 	private final MqttConnectionOptions connectionOptions;
 
@@ -301,12 +302,16 @@ protected void doStop() {
 					this.readyToSubscribeOnStart = true;
 
 				}
-				if (getClientManager() == null) {
+				ClientManager<IMqttAsyncClient, MqttConnectionOptions> clientManager = getClientManager();
+				if (clientManager == null) {
 					this.mqttClient.disconnectForcibly(getDisconnectCompletionTimeout());
 					if (getConnectionInfo().isAutomaticReconnect()) {
 						MqttUtils.stopClientReconnectCycle(this.mqttClient);
 					}
 				}
+				else {
+					clientManager.removeDefaultMessageHandler(this.defaultMessageHandler);
+				}
 			}
 		}
 		catch (MqttException ex) {
@@ -325,10 +330,6 @@ private void unsubscribe(String... topics) throws MqttException {
 		catch (ConcurrentModificationException ex) {
 			logger.error(ex, () -> "Error unsubscribing from " + Arrays.toString(topics));
 		}
-		ClientManager<IMqttAsyncClient, MqttConnectionOptions> clientManager = getClientManager();
-		if (clientManager != null) {
-			clientManager.removeDefaultMessageHandler(this.defaultMessageHandler);
-		}
 	}
 
 	@Override
@@ -380,10 +381,10 @@ public void addTopic(String topic, int qos) {
 	public void removeTopic(String... topic) {
 		this.topicLock.lock();
 		try {
+			super.removeTopic(topic);
 			if (this.mqttClient != null && this.mqttClient.isConnected()) {
 				unsubscribe(topic);
 			}
-			super.removeTopic(topic);
 			if (!CollectionUtils.isEmpty(this.subscriptions)) {
 				this.subscriptions.removeIf((sub) -> ObjectUtils.containsElement(topic, sub.getTopic()));
 			}
@@ -398,12 +399,6 @@ public void removeTopic(String... topic) {
 
 	@Override
 	public void messageArrived(String topic, MqttMessage mqttMessage) {
-		if (!isTopicSubscribed(topic)) {
-			logger.trace(() ->
-					"Arrived message on topic '" + topic + "' this channel adapter is not subscribed to. Ignoring...");
-			return;
-		}
-
 		Map<String, Object> headers = this.headerMapper.toHeaders(mqttMessage.getProperties());
 		headers.put(MqttHeaders.ID, mqttMessage.getId());
 		headers.put(MqttHeaders.RECEIVED_QOS, mqttMessage.getQos());
@@ -438,17 +433,6 @@ public void messageArrived(String topic, MqttMessage mqttMessage) {
 		}
 	}
 
-	private boolean isTopicSubscribed(String receivedTopic) {
-		for (String subscribedTopic : getTopic()) {
-			if (subscribedTopic.equals(receivedTopic)
-					|| subscribedTopic.startsWith("$share/") && subscribedTopic.endsWith(receivedTopic)) {
-
-				return true;
-			}
-		}
-		return false;
-	}
-
 	@Override
 	public void disconnected(MqttDisconnectResponse disconnectResponse) {
 		if (isRunning()) {
@@ -550,6 +534,20 @@ public void authPacketArrived(int reasonCode, MqttProperties properties) {
 
 	}
 
+	private void messageArrivedIfMatched(String topic, MqttMessage mqttMessage) {
+		for (String subscribedTopic : getTopic()) {
+			if (subscribedTopic.startsWith("$share/")) {
+				subscribedTopic = subscribedTopic.split("/", 3)[2];
+			}
+			if (MqttTopicValidator.isMatched(subscribedTopic, topic)) {
+				messageArrived(topic, mqttMessage);
+				return;
+			}
+		}
+		logger.trace(() ->
+				"Arrived message on topic '" + topic + "' this channel adapter is not subscribed to. Ignoring...");
+	}
+
 	private static String obtainServerUrlFromOptions(MqttConnectionOptions connectionOptions) {
 		Assert.notNull(connectionOptions, "'connectionOptions' must not be null");
 		String[] serverURIs = connectionOptions.getServerURIs();

spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/Mqttv5BackToBackTests.java

@@ -139,7 +139,7 @@ public void testSimpleMqttv5Interaction() {
 
 	@Test
 	public void testSharedTopicMqttv5Interaction() {
-		this.mqttv5MessageDrivenChannelAdapter.addTopic("$share/group/testTopic");
+		this.mqttv5MessageDrivenChannelAdapter.addTopic("$share/group/#");
 
 		String testPayload = "shared topic payload";
 		this.mqttOutFlowInput.send(