Add Inbound Channel Adapter for MQTT based on hivemq

build.gradle

@@ -596,6 +596,13 @@ project('spring-integration-hazelcast') {
     }
 }
 
+project('spring-integration-hivemq') {
+	description = 'Spring Integration Hivemq MQTT Support'
+	dependencies {
+		api libs.com.hivemq.mqtt.client
+	}
+}
+
 project('spring-integration-http') {
     description = 'Spring Integration HTTP Support'
     dependencies {

gradle/libs.versions.toml

@@ -23,6 +23,7 @@ h2Version = "2.4.240"
 hamcrestVersion = "3.0"
 hazelcastVersion = "5.6.0"
 hibernateVersion = "7.3.2.Final"
+hivemqMqttClientVersion="1.3.14"
 hsqldbVersion = "2.7.4"
 jackson3Version = "3.1.3"
 jacksonVersion = "2.21.3"
@@ -90,6 +91,7 @@ com-github-spotbugs-annotations = { module = 'com.github.spotbugs:spotbugs-annot
 com-google-protobuf-bom = { module = "com.google.protobuf:protobuf-bom", version.ref = "protobufVersion" }
 com-h2database-h2 = { module = "com.h2database:h2", version.ref = "h2Version" }
 com-hazelcast = { module = "com.hazelcast:hazelcast", version.ref = "hazelcastVersion" }
+com-hivemq-mqtt-client = { module="com.hivemq:hivemq-mqtt-client", version.ref = "hivemqMqttClientVersion"}
 com-icegreen-greenmail = { module = "com.icegreen:greenmail", version.ref = "greenmailVersion" }
 com-jayway-jsonpath = { module = "com.jayway.jsonpath:json-path", version.ref = "jsonpathVersion" }
 com-mysql-connector = { module = "com.mysql:mysql-connector-j", version.ref = "mysqlVersion" }

spring-integration-hivemq/src/main/java/com/springframework/integration/hivemq/MqttClientConnectionCoordinator.java

@@ -0,0 +1,52 @@
+/*
+ * Copyright 2026-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.springframework.integration.hivemq;
+
+import java.util.concurrent.CompletableFuture;
+
+import org.jspecify.annotations.Nullable;
+
+/**
+ * An abstraction for coordinating MQTT client connections and disconnections, ensures thread-safe operations.
+ *
+ * @param <T> MQTT Client type
+ * @param <C> MQTT Connect Message type
+ * @param <D> MQTT Disconnect Message type
+ *
+ * @author Jiandong Ma
+ *
+ * @since 7.2
+ */
+public interface MqttClientConnectionCoordinator<T, C, D> {
+
+	/**
+	 * Connect to MQTT broker using the mqttClient and mqttConnect.
+	 * @param mqttClient the mqttClient
+	 * @param mqttConnect the mqttConnect
+	 * @return CompletableFuture
+	 */
+	CompletableFuture<?> connect(T mqttClient, C mqttConnect);
+
+	/**
+	 * Disconnect from MQTT broker using the mqttClient and mqttDisconnect.
+	 * @param mqttClient the mqttClient
+	 * @param mqttDisconnect the mqttDisconnect, for MQTT v3, it is always null.
+	 * @return CompletableFuture
+	 */
+	CompletableFuture<Void> disconnect(T mqttClient, @Nullable D mqttDisconnect);
+
+}

spring-integration-hivemq/src/main/java/com/springframework/integration/hivemq/MqttClientConnectionCoordinators.java

@@ -0,0 +1,122 @@
+/*
+ * Copyright 2026-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.springframework.integration.hivemq;
+
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+
+import com.hivemq.client.internal.mqtt.message.connect.mqtt3.Mqtt3ConnectView;
+import com.hivemq.client.mqtt.mqtt3.Mqtt3AsyncClient;
+import com.hivemq.client.mqtt.mqtt3.message.connect.connack.Mqtt3ConnAck;
+import com.hivemq.client.mqtt.mqtt5.Mqtt5AsyncClient;
+import com.hivemq.client.mqtt.mqtt5.message.connect.Mqtt5Connect;
+import com.hivemq.client.mqtt.mqtt5.message.connect.connack.Mqtt5ConnAck;
+import com.hivemq.client.mqtt.mqtt5.message.disconnect.Mqtt5Disconnect;
+import org.jspecify.annotations.Nullable;
+
+/**
+ * The {@link MqttClientConnectionCoordinators} for getting a particular {@link MqttClientConnectionCoordinator}.
+ *
+ * @author Jiandong Ma
+ *
+ * @since 7.2
+ */
+public final class MqttClientConnectionCoordinators {
+
+	private static final Mqtt5AsyncClientConnectionCoordinator MQTT_5_ASYNC_CLIENT_CONNECTION_COORDINATOR
+			= new Mqtt5AsyncClientConnectionCoordinator();
+
+	private static final Mqtt3AsyncClientConnectionCoordinator MQTT_3_ASYNC_CLIENT_CONNECTION_COORDINATOR
+			= new Mqtt3AsyncClientConnectionCoordinator();
+
+	public static MqttClientConnectionCoordinator<Mqtt5AsyncClient, Mqtt5Connect, Mqtt5Disconnect> mqtt5AsyncClient() {
+		return MQTT_5_ASYNC_CLIENT_CONNECTION_COORDINATOR;
+	}
+
+	public static MqttClientConnectionCoordinator<Mqtt3AsyncClient, Mqtt3ConnectView, Object> mqtt3AsyncClient() {
+		return MQTT_3_ASYNC_CLIENT_CONNECTION_COORDINATOR;
+	}
+
+	private MqttClientConnectionCoordinators() {
+	}
+
+	/**
+	 * A {@link MqttClientConnectionCoordinator} implementation for {@link Mqtt5AsyncClient}.
+	 */
+	static class Mqtt5AsyncClientConnectionCoordinator
+			implements MqttClientConnectionCoordinator<Mqtt5AsyncClient, Mqtt5Connect, Mqtt5Disconnect> {
+
+		Map<Mqtt5AsyncClient, CompletableFuture<Mqtt5ConnAck>> CONNECT_FUTURE_MAP = new ConcurrentHashMap<>();
+
+		Map<Mqtt5AsyncClient, CompletableFuture<Void>> DISCONNECT_FUTURE_MAP = new ConcurrentHashMap<>();
+
+		@Override
+		public CompletableFuture<Mqtt5ConnAck> connect(Mqtt5AsyncClient mqttClient, Mqtt5Connect mqttConnect) {
+			return this.CONNECT_FUTURE_MAP.computeIfAbsent(mqttClient, client -> {
+				// Remove from disconnect map, in case dirty cache between lifecycle methods
+				this.DISCONNECT_FUTURE_MAP.remove(client);
+
+				return client.connect(mqttConnect);
+			});
+		}
+
+		@Override
+		public CompletableFuture<Void> disconnect(Mqtt5AsyncClient mqttClient, @Nullable Mqtt5Disconnect mqttDisconnect) {
+			return this.DISCONNECT_FUTURE_MAP.computeIfAbsent(mqttClient, client -> {
+				// Remove from connect map,  in case dirty cache between lifecycle methods
+				this.CONNECT_FUTURE_MAP.remove(client);
+
+				return mqttDisconnect != null ? client.disconnect(mqttDisconnect) : client.disconnect();
+			});
+		}
+
+	}
+
+	/**
+	 * A {@link MqttClientConnectionCoordinator} implementation for {@link Mqtt3AsyncClient}.
+	 */
+	static class Mqtt3AsyncClientConnectionCoordinator
+			implements MqttClientConnectionCoordinator<Mqtt3AsyncClient, Mqtt3ConnectView, Object> {
+
+		Map<Mqtt3AsyncClient, CompletableFuture<Mqtt3ConnAck>> CONNECT_FUTURE_MAP = new ConcurrentHashMap<>();
+
+		Map<Mqtt3AsyncClient, CompletableFuture<Void>> DISCONNECT_FUTURE_MAP = new ConcurrentHashMap<>();
+
+		@Override
+		public CompletableFuture<Mqtt3ConnAck> connect(Mqtt3AsyncClient mqttClient, Mqtt3ConnectView mqttConnect) {
+			return this.CONNECT_FUTURE_MAP.computeIfAbsent(mqttClient, client -> {
+				// Remove from disconnect map, in case dirty cache between lifecycle methods
+				this.DISCONNECT_FUTURE_MAP.remove(client);
+
+				return client.connect(mqttConnect);
+			});
+		}
+
+		@Override
+		public CompletableFuture<Void> disconnect(Mqtt3AsyncClient mqttClient, @Nullable Object mqttDisconnect) {
+			return this.DISCONNECT_FUTURE_MAP.computeIfAbsent(mqttClient, client -> {
+				// Remove from connect map,  in case dirty cache between lifecycle methods
+				this.CONNECT_FUTURE_MAP.remove(client);
+
+				return client.disconnect();
+			});
+		}
+
+	}
+
+}

spring-integration-hivemq/src/main/java/com/springframework/integration/hivemq/event/MqttConnectionFailedEvent.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2026-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.springframework.integration.hivemq.event;
+
+import org.jspecify.annotations.Nullable;
+
+/**
+ * The {@link MqttIntegrationEvent} to notify about lost connection to the server.
+ * When normal disconnection is happened (initiated by the server), the {@code cause} is null.
+ *
+ * @author Gary Russell
+ * @author Artem Bilan
+ *
+ * @since 7.2
+ */
+@SuppressWarnings("serial")
+public class MqttConnectionFailedEvent extends MqttIntegrationEvent {
+
+	public MqttConnectionFailedEvent(Object source) {
+		super(source);
+	}
+
+	public MqttConnectionFailedEvent(Object source, @Nullable Throwable cause) {
+		super(source, cause);
+	}
+
+}

spring-integration-hivemq/src/main/java/com/springframework/integration/hivemq/event/MqttIntegrationEvent.java

@@ -0,0 +1,42 @@
+/*
+ * Copyright 2026-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.springframework.integration.hivemq.event;
+
+import org.jspecify.annotations.Nullable;
+
+import org.springframework.integration.events.IntegrationEvent;
+
+/**
+ * Base class for Mqtt Events.
+ *
+ * @author Gary Russell
+ * @author Artem Bilan
+ *
+ * @since 7.2
+ */
+@SuppressWarnings("serial")
+public abstract class MqttIntegrationEvent extends IntegrationEvent {
+
+	public MqttIntegrationEvent(Object source) {
+		super(source);
+	}
+
+	public MqttIntegrationEvent(Object source, @Nullable Throwable cause) {
+		super(source, cause);
+	}
+
+}

spring-integration-hivemq/src/main/java/com/springframework/integration/hivemq/event/MqttSubscribedEvent.java

@@ -0,0 +1,43 @@
+/*
+ * Copyright 2026-present the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.springframework.integration.hivemq.event;
+
+/**
+ * @author Gary Russell
+ *
+ * @since 7.2
+ */
+@SuppressWarnings("serial")
+public class MqttSubscribedEvent extends MqttIntegrationEvent {
+
+	private final String message;
+
+	public MqttSubscribedEvent(Object source, String message) {
+		super(source);
+		this.message = message;
+	}
+
+	public String getMessage() {
+		return this.message;
+	}
+
+	@Override
+	public String toString() {
+		return "MqttSubscribedEvent [message=" + this.message + ", source=" + source + "]";
+	}
+
+}

spring-integration-hivemq/src/main/java/com/springframework/integration/hivemq/event/package-info.java

@@ -0,0 +1,5 @@
+/**
+ * ApplicationEvents generated by the hivemq MQTT module.
+ */
+@org.jspecify.annotations.NullMarked
+package com.springframework.integration.hivemq.event;