diff --git a/build.gradle b/build.gradle index 1b94b592d11..880f7888014 100644 --- a/build.gradle +++ b/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 { diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml index c5e8d431cd6..616bb8a7114 100644 --- a/gradle/libs.versions.toml +++ b/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" } diff --git a/spring-integration-hivemq/src/main/java/com/springframework/integration/hivemq/MqttClientConnectionCoordinator.java b/spring-integration-hivemq/src/main/java/com/springframework/integration/hivemq/MqttClientConnectionCoordinator.java new file mode 100644 index 00000000000..ccef6ea31f8 --- /dev/null +++ b/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 MQTT Client type + * @param MQTT Connect Message type + * @param MQTT Disconnect Message type + * + * @author Jiandong Ma + * + * @since 7.2 + */ +public interface MqttClientConnectionCoordinator { + + /** + * 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 disconnect(T mqttClient, @Nullable D mqttDisconnect); + +} diff --git a/spring-integration-hivemq/src/main/java/com/springframework/integration/hivemq/MqttClientConnectionCoordinators.java b/spring-integration-hivemq/src/main/java/com/springframework/integration/hivemq/MqttClientConnectionCoordinators.java new file mode 100644 index 00000000000..1049fd252f3 --- /dev/null +++ b/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() { + return MQTT_5_ASYNC_CLIENT_CONNECTION_COORDINATOR; + } + + public static MqttClientConnectionCoordinator mqtt3AsyncClient() { + return MQTT_3_ASYNC_CLIENT_CONNECTION_COORDINATOR; + } + + private MqttClientConnectionCoordinators() { + } + + /** + * A {@link MqttClientConnectionCoordinator} implementation for {@link Mqtt5AsyncClient}. + */ + static class Mqtt5AsyncClientConnectionCoordinator + implements MqttClientConnectionCoordinator { + + Map> CONNECT_FUTURE_MAP = new ConcurrentHashMap<>(); + + Map> DISCONNECT_FUTURE_MAP = new ConcurrentHashMap<>(); + + @Override + public CompletableFuture 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 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 { + + Map> CONNECT_FUTURE_MAP = new ConcurrentHashMap<>(); + + Map> DISCONNECT_FUTURE_MAP = new ConcurrentHashMap<>(); + + @Override + public CompletableFuture 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 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(); + }); + } + + } + +} diff --git a/spring-integration-hivemq/src/main/java/com/springframework/integration/hivemq/event/MqttConnectionFailedEvent.java b/spring-integration-hivemq/src/main/java/com/springframework/integration/hivemq/event/MqttConnectionFailedEvent.java new file mode 100644 index 00000000000..bff1d75486f --- /dev/null +++ b/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); + } + +} diff --git a/spring-integration-hivemq/src/main/java/com/springframework/integration/hivemq/event/MqttIntegrationEvent.java b/spring-integration-hivemq/src/main/java/com/springframework/integration/hivemq/event/MqttIntegrationEvent.java new file mode 100644 index 00000000000..df4d35e53ff --- /dev/null +++ b/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); + } + +} diff --git a/spring-integration-hivemq/src/main/java/com/springframework/integration/hivemq/event/MqttSubscribedEvent.java b/spring-integration-hivemq/src/main/java/com/springframework/integration/hivemq/event/MqttSubscribedEvent.java new file mode 100644 index 00000000000..fc973e493a4 --- /dev/null +++ b/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 + "]"; + } + +} diff --git a/spring-integration-hivemq/src/main/java/com/springframework/integration/hivemq/event/package-info.java b/spring-integration-hivemq/src/main/java/com/springframework/integration/hivemq/event/package-info.java new file mode 100644 index 00000000000..cdaef53d6d2 --- /dev/null +++ b/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; diff --git a/spring-integration-hivemq/src/main/java/com/springframework/integration/hivemq/inbound/AbstractMqttMessageDrivenChannelAdapter.java b/spring-integration-hivemq/src/main/java/com/springframework/integration/hivemq/inbound/AbstractMqttMessageDrivenChannelAdapter.java new file mode 100644 index 00000000000..26ede7901ac --- /dev/null +++ b/spring-integration-hivemq/src/main/java/com/springframework/integration/hivemq/inbound/AbstractMqttMessageDrivenChannelAdapter.java @@ -0,0 +1,139 @@ +/* + * 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.inbound; + +import java.util.concurrent.Executor; + +import com.hivemq.client.mqtt.MqttClient; +import com.hivemq.client.mqtt.datatypes.MqttQos; +import org.jspecify.annotations.Nullable; + +import org.springframework.context.ApplicationEventPublisher; +import org.springframework.context.ApplicationEventPublisherAware; +import org.springframework.integration.context.IntegrationContextUtils; +import org.springframework.integration.endpoint.MessageProducerSupport; +import org.springframework.messaging.converter.SmartMessageConverter; +import org.springframework.util.Assert; + +/** + * Abstract class for MQTT Message-Driven Channel Adapters. + * + * @param MQTT Client type + * + * @author Jiandong Ma + * + * @since 7.2 + */ +public abstract class AbstractMqttMessageDrivenChannelAdapter + extends MessageProducerSupport implements ApplicationEventPublisherAware { + + @SuppressWarnings("NullAway.Init") + protected ApplicationEventPublisher applicationEventPublisher; + + @SuppressWarnings("NullAway.Init") + protected SmartMessageConverter messageConverter; + + protected final T mqttClient; + + protected final String topic; + + protected MqttQos qos = MqttQos.AT_LEAST_ONCE; + + protected boolean manualAck = false; + + protected @Nullable Executor executor; + + protected Class payloadType = byte[].class; + + protected AbstractMqttMessageDrivenChannelAdapter(T mqttClient, String topic) { + Assert.notNull(mqttClient, "'mqttClient' cannot be null"); + Assert.hasText(topic, "The topic to subscribe cannot be empty string"); + this.mqttClient = mqttClient; + this.topic = topic; + + if (mqttClient.getConfig().getAutomaticReconnect().isEmpty()) { + logger.warn("it is recommended to enable 'automaticReconnect' when set this `mqttClient`. " + + "Otherwise connection check and reconnect should be done manually."); + } + } + + @Override + public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) { + this.applicationEventPublisher = applicationEventPublisher; + } + + @Override + protected void onInit() { + super.onInit(); + if (this.messageConverter == null) { + String messageConverterBeanName = IntegrationContextUtils.ARGUMENT_RESOLVER_MESSAGE_CONVERTER_BEAN_NAME; + setMessageConverter(getBeanFactory().getBean(messageConverterBeanName, SmartMessageConverter.class)); + } + } + + /** + * Set the QoS for the topic + * @param qos The qos value + */ + public void setQos(MqttQos qos) { + Assert.notNull(qos, "'qos' must not be null."); + this.qos = qos; + } + + /** + * Set whether the Publish messages are acknowledged manually. + * @param manualAck true for manual ack. + */ + public void setManualAcknowledgement(boolean manualAck) { + this.manualAck = manualAck; + } + + /** + * Set the executor where the message callback is executed on. + * @param executor the executor. + */ + public void setExecutor(Executor executor) { + Assert.notNull(executor, "'executor' must not be null."); + this.executor = executor; + } + + /** + * Set the type of the target message payload to produce after conversion from MQTT message. + * Defaults to {@code byte[].class}. Can be set to {@code Mqtt5Publish} for v5 or {@code Mqtt3Publish} + * to produce the whole MQTT message as a payload. + * @param payloadType the expected payload type to convert MQTT message to. + */ + public void setPayloadType(Class payloadType) { + Assert.notNull(payloadType, "'payloadType' must not be null."); + this.payloadType = payloadType; + } + + /** + * Set the messageConverter to convert the payload to the expected payloadType. + * @param messageConverter the messageConverter + */ + public void setMessageConverter(SmartMessageConverter messageConverter) { + Assert.notNull(messageConverter, "'messageConverter' must not be null."); + this.messageConverter = messageConverter; + } + + @Override + public String getComponentType() { + return "mqtt:inbound-channel-adapter"; + } + +} diff --git a/spring-integration-hivemq/src/main/java/com/springframework/integration/hivemq/inbound/Mqtt3MessageDrivenChannelAdapter.java b/spring-integration-hivemq/src/main/java/com/springframework/integration/hivemq/inbound/Mqtt3MessageDrivenChannelAdapter.java new file mode 100644 index 00000000000..0f82c9d78b4 --- /dev/null +++ b/spring-integration-hivemq/src/main/java/com/springframework/integration/hivemq/inbound/Mqtt3MessageDrivenChannelAdapter.java @@ -0,0 +1,145 @@ +/* + * 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.inbound; + +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.CompletableFuture; + +import com.hivemq.client.internal.mqtt.message.connect.mqtt3.Mqtt3ConnectView; +import com.hivemq.client.mqtt.mqtt3.Mqtt3AsyncClient; +import com.hivemq.client.mqtt.mqtt3.message.publish.Mqtt3Publish; +import com.hivemq.client.mqtt.mqtt3.message.subscribe.Mqtt3Subscribe; +import com.hivemq.client.mqtt.mqtt3.message.subscribe.suback.Mqtt3SubAck; +import com.springframework.integration.hivemq.MqttClientConnectionCoordinators; +import com.springframework.integration.hivemq.event.MqttConnectionFailedEvent; +import com.springframework.integration.hivemq.event.MqttSubscribedEvent; +import com.springframework.integration.hivemq.support.MqttHeaders; + +import org.springframework.integration.IntegrationMessageHeaderAccessor; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHeaders; +import org.springframework.messaging.support.GenericMessage; +import org.springframework.util.Assert; + +/** + * The {@link AbstractMqttMessageDrivenChannelAdapter} implementation for MQTT v3. + * + * @author Jiandong Ma + * + * @since 7.2 + */ +public class Mqtt3MessageDrivenChannelAdapter extends + AbstractMqttMessageDrivenChannelAdapter { + + private Mqtt3ConnectView mqtt3ConnectView = Mqtt3ConnectView.DEFAULT; + + public Mqtt3MessageDrivenChannelAdapter(Mqtt3AsyncClient mqttClient, String topic) { + super(mqttClient, topic); + } + + /** + * Set the Connect message. + * @param mqtt3ConnectView the mqtt3ConnectView + */ + public void setMqtt3ConnectView(Mqtt3ConnectView mqtt3ConnectView) { + Assert.notNull(mqtt3ConnectView, "'mqtt3ConnectView' must not be null"); + this.mqtt3ConnectView = mqtt3ConnectView; + } + + @Override + protected void onInit() { + super.onInit(); + } + + @Override + protected void doStart() { + super.doStart(); + MqttClientConnectionCoordinators.mqtt3AsyncClient().connect(mqttClient, this.mqtt3ConnectView) + .whenComplete((connAck, throwable) -> { + if (throwable != null) { + applicationEventPublisher.publishEvent(new MqttConnectionFailedEvent(this, throwable)); + logger.error(throwable, "MQTT client failed to connect. " + mqttClient); + } + else { + subscribe(); + } + }); + } + + @Override + protected void doStop() { + super.doStop(); + MqttClientConnectionCoordinators.mqtt3AsyncClient().disconnect(mqttClient, null) + .whenComplete((unused, throwable) -> { + if (throwable != null) { + logger.error(throwable, "MQTT client failed to disconnect." + mqttClient); + } + }); + } + + private void subscribe() { + Mqtt3Subscribe mqtt3Subscribe = Mqtt3Subscribe.builder() + .topicFilter(topic) + .qos(qos) + .build(); + CompletableFuture subscribeFuture; + if (executor != null) { + subscribeFuture = mqttClient.subscribe(mqtt3Subscribe, this::messageListener, executor, manualAck); + } + else { + subscribeFuture = mqttClient.subscribe(mqtt3Subscribe, this::messageListener, manualAck); + } + subscribeFuture.whenComplete((subAck, throwable) -> { + if (throwable != null) { + logger.error(throwable, "MQTT client failed to subscribe topic : " + topic); + applicationEventPublisher.publishEvent(new MqttConnectionFailedEvent(this, throwable)); + } + else { + String msg = "MQTT client subscribe topic: " + topic; + logger.debug(msg); + applicationEventPublisher.publishEvent(new MqttSubscribedEvent(this, msg)); + } + }); + } + + private void messageListener(Mqtt3Publish mqttMessage) { + Map headers = new HashMap<>(); + headers.put(MqttHeaders.RECEIVED_QOS, mqttMessage.getQos()); + headers.put(MqttHeaders.RECEIVED_RETAINED, mqttMessage.isRetain()); + headers.put(MqttHeaders.RECEIVED_TOPIC, mqttMessage.getTopic().toString()); + + if (manualAck) { + headers.put(IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK, mqttMessage); + } + + Object payload = Mqtt3Publish.class.isAssignableFrom(this.payloadType) + ? mqttMessage + : mqttMessage.getPayloadAsBytes(); + + Message message; + if (Mqtt3Publish.class.isAssignableFrom(this.payloadType) || byte[].class.isAssignableFrom(this.payloadType)) { + message = new GenericMessage<>(payload, headers); + } + else { + message = this.messageConverter.toMessage(payload, new MessageHeaders(headers), this.payloadType); + } + + sendMessage(message); + } + +} diff --git a/spring-integration-hivemq/src/main/java/com/springframework/integration/hivemq/inbound/Mqtt5MessageDrivenChannelAdapter.java b/spring-integration-hivemq/src/main/java/com/springframework/integration/hivemq/inbound/Mqtt5MessageDrivenChannelAdapter.java new file mode 100644 index 00000000000..c81e3e33c6e --- /dev/null +++ b/spring-integration-hivemq/src/main/java/com/springframework/integration/hivemq/inbound/Mqtt5MessageDrivenChannelAdapter.java @@ -0,0 +1,211 @@ +/* + * 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.inbound; + +import java.util.Map; +import java.util.concurrent.CompletableFuture; + +import com.hivemq.client.internal.mqtt.message.connect.MqttConnect; +import com.hivemq.client.internal.mqtt.message.disconnect.MqttDisconnect; +import com.hivemq.client.internal.mqtt.message.subscribe.MqttSubscription; +import com.hivemq.client.mqtt.mqtt5.Mqtt5AsyncClient; +import com.hivemq.client.mqtt.mqtt5.message.publish.Mqtt5Publish; +import com.hivemq.client.mqtt.mqtt5.message.subscribe.Mqtt5RetainHandling; +import com.hivemq.client.mqtt.mqtt5.message.subscribe.Mqtt5Subscribe; +import com.hivemq.client.mqtt.mqtt5.message.subscribe.suback.Mqtt5SubAck; +import com.springframework.integration.hivemq.MqttClientConnectionCoordinators; +import com.springframework.integration.hivemq.event.MqttConnectionFailedEvent; +import com.springframework.integration.hivemq.event.MqttSubscribedEvent; +import com.springframework.integration.hivemq.support.Mqtt5HeaderMapper; +import com.springframework.integration.hivemq.support.MqttHeaders; + +import org.springframework.integration.IntegrationMessageHeaderAccessor; +import org.springframework.integration.mapping.HeaderMapper; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHeaders; +import org.springframework.messaging.support.GenericMessage; +import org.springframework.util.Assert; + +/** + * The {@link AbstractMqttMessageDrivenChannelAdapter} implementation for MQTT v5. + * + * @author Jiandong Ma + * + * @since 7.2 + */ +public class Mqtt5MessageDrivenChannelAdapter extends + AbstractMqttMessageDrivenChannelAdapter { + + private HeaderMapper headerMapper = new Mqtt5HeaderMapper(); + + private MqttConnect mqttConnect = MqttConnect.DEFAULT; + + private MqttDisconnect mqttDisConnect = MqttDisconnect.DEFAULT; + + // [Start] Additional MQTT v5 subscription Options + + private boolean noLocal = MqttSubscription.DEFAULT_NO_LOCAL; + + private Mqtt5RetainHandling retainHandling = MqttSubscription.DEFAULT_RETAIN_HANDLING; + + private boolean retainAsPublished = MqttSubscription.DEFAULT_RETAIN_AS_PUBLISHED; + + // [End] + + public Mqtt5MessageDrivenChannelAdapter(Mqtt5AsyncClient mqttClient, String topic) { + super(mqttClient, topic); + } + + @Override + protected void onInit() { + super.onInit(); + } + + /** + * Set the HeaderMapper to map the {@code Mqtt5Publish} optional data + * @param headerMapper the headMapper + */ + public void setHeaderMapper(HeaderMapper headerMapper) { + Assert.notNull(headerMapper, "'headerMapper' must not be null."); + this.headerMapper = headerMapper; + } + + /** + * Set the Connect message. + * @param mqttConnect the mqttConnect + */ + public void setMqttConnect(MqttConnect mqttConnect) { + Assert.notNull(mqttConnect, "'mqttConnect' must not be null."); + this.mqttConnect = mqttConnect; + } + + /** + * Set whether the client must not receive messages published by itself. + * @param noLocal whether the subscription is not local. + */ + public void setNoLocal(boolean noLocal) { + this.noLocal = noLocal; + } + + /** + * Set the handling of retained message for this Subscription. + * @param retainHandling the retain handling. + */ + public void setRetainHandling(Mqtt5RetainHandling retainHandling) { + Assert.notNull(retainHandling, "'retainHandling' must not be null."); + this.retainHandling = retainHandling; + } + + /** + * Set whether the retain flag for incoming Publish messages must be set to its original value. + * @param retainAsPublished the retainAsPublished. + */ + public void setRetainAsPublished(boolean retainAsPublished) { + this.retainAsPublished = retainAsPublished; + } + + /** + * Set the Disconnect message. + * @param mqttDisconnect the mqttDisconnect + */ + public void setMqttDisconnect(MqttDisconnect mqttDisconnect) { + Assert.notNull(mqttDisconnect, "'mqttDisconnect' must not be null."); + this.mqttDisConnect = mqttDisconnect; + } + + @Override + protected void doStart() { + super.doStart(); + MqttClientConnectionCoordinators.mqtt5AsyncClient().connect(mqttClient, this.mqttConnect) + .whenComplete((connAck, throwable) -> { + if (throwable != null) { + MqttConnectionFailedEvent event = new MqttConnectionFailedEvent(this, throwable); + applicationEventPublisher.publishEvent(event); + logger.error(throwable, "MQTT client failed to connect. " + throwable.getMessage()); + } + else { + subscribe(); + } + }); + } + + @Override + protected void doStop() { + super.doStop(); + MqttClientConnectionCoordinators.mqtt5AsyncClient().disconnect(mqttClient, this.mqttDisConnect) + .whenComplete((unused, throwable) -> { + if (throwable != null) { + logger.error(throwable, "MQTT client failed to disconnect. " + throwable.getMessage()); + } + }); + } + + private void subscribe() { + Mqtt5Subscribe mqtt5Subscribe = Mqtt5Subscribe.builder() + .topicFilter(topic) + .qos(qos) + .noLocal(this.noLocal) + .retainHandling(this.retainHandling) + .retainAsPublished(this.retainAsPublished) + .build(); + CompletableFuture subscribeFuture; + if (executor != null) { + subscribeFuture = mqttClient.subscribe(mqtt5Subscribe, this::messageListener, executor, manualAck); + } + else { + subscribeFuture = mqttClient.subscribe(mqtt5Subscribe, this::messageListener, manualAck); + } + subscribeFuture.whenComplete((subAck, throwable) -> { + if (throwable != null) { + logger.error(throwable, "MQTT client failed to subscribe topic: " + topic); + applicationEventPublisher.publishEvent(new MqttConnectionFailedEvent(this, throwable)); + } + else { + String msg = "MQTT client subscribe topic: " + topic; + logger.debug(msg); + applicationEventPublisher.publishEvent(new MqttSubscribedEvent(this, msg)); + } + }); + } + + private void messageListener(Mqtt5Publish mqtt5Publish) { + Map headers = this.headerMapper.toHeaders(mqtt5Publish); + + headers.put(MqttHeaders.RECEIVED_QOS, mqtt5Publish.getQos()); + headers.put(MqttHeaders.RECEIVED_RETAINED, mqtt5Publish.isRetain()); + headers.put(MqttHeaders.RECEIVED_TOPIC, mqtt5Publish.getTopic().toString()); + + if (manualAck) { + headers.put(IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK, mqtt5Publish); + } + + Object payload = Mqtt5Publish.class.isAssignableFrom(this.payloadType) + ? mqtt5Publish + : mqtt5Publish.getPayloadAsBytes(); + + Message message; + if (Mqtt5Publish.class.isAssignableFrom(this.payloadType) || byte[].class.isAssignableFrom(this.payloadType)) { + message = new GenericMessage<>(payload, headers); + } + else { + message = this.messageConverter.toMessage(payload, new MessageHeaders(headers), this.payloadType); + } + + sendMessage(message); + } + +} diff --git a/spring-integration-hivemq/src/main/java/com/springframework/integration/hivemq/inbound/package-info.java b/spring-integration-hivemq/src/main/java/com/springframework/integration/hivemq/inbound/package-info.java new file mode 100644 index 00000000000..b61abf0acbf --- /dev/null +++ b/spring-integration-hivemq/src/main/java/com/springframework/integration/hivemq/inbound/package-info.java @@ -0,0 +1,5 @@ +/** + * Provides inbound MQTT adapter components. + */ +@org.jspecify.annotations.NullMarked +package com.springframework.integration.hivemq.inbound; \ No newline at end of file diff --git a/spring-integration-hivemq/src/main/java/com/springframework/integration/hivemq/package-info.java b/spring-integration-hivemq/src/main/java/com/springframework/integration/hivemq/package-info.java new file mode 100644 index 00000000000..c44715aef43 --- /dev/null +++ b/spring-integration-hivemq/src/main/java/com/springframework/integration/hivemq/package-info.java @@ -0,0 +1,5 @@ +/** + * Root package of the MQTT adapter module. + */ +@org.jspecify.annotations.NullMarked +package com.springframework.integration.hivemq; \ No newline at end of file diff --git a/spring-integration-hivemq/src/main/java/com/springframework/integration/hivemq/support/Mqtt5HeaderMapper.java b/spring-integration-hivemq/src/main/java/com/springframework/integration/hivemq/support/Mqtt5HeaderMapper.java new file mode 100644 index 00000000000..f180d8fcea3 --- /dev/null +++ b/spring-integration-hivemq/src/main/java/com/springframework/integration/hivemq/support/Mqtt5HeaderMapper.java @@ -0,0 +1,89 @@ +/* + * 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.support; + +import java.util.HashMap; +import java.util.Map; +import java.util.Objects; + +import com.hivemq.client.mqtt.mqtt5.datatypes.Mqtt5UserProperty; +import com.hivemq.client.mqtt.mqtt5.message.publish.Mqtt5Publish; +import org.jspecify.annotations.Nullable; + +import org.springframework.core.log.LogAccessor; +import org.springframework.core.log.LogMessage; +import org.springframework.integration.mapping.HeaderMapper; +import org.springframework.messaging.MessageHeaders; +import org.springframework.util.PatternMatchUtils; + +/** + * The default {@link HeaderMapper} implementation for MQTT v5 message properties mapping. + * + * @author Artem Bilan + * @author Jiandong Ma + * + * @since 7.2 + */ +public class Mqtt5HeaderMapper implements HeaderMapper { + + private static final LogAccessor LOGGER = new LogAccessor(Mqtt5HeaderMapper.class); + + private final String[] inboundHeaderNames = {"*"}; + + @Override + public void fromHeaders(MessageHeaders headers, Mqtt5Publish target) { + + } + + @Override + public Map toHeaders(Mqtt5Publish mqtt5Publish) { + Map headers = new HashMap<>(); + + headers.compute(MessageHeaders.CONTENT_TYPE, (k, v) -> + mapPropertyIfMatch(k, mqtt5Publish.getContentType().map(Object::toString).orElse(null))); + headers.compute(MqttHeaders.RESPONSE_TOPIC, (k, v) -> + mapPropertyIfMatch(k, mqtt5Publish.getResponseTopic().map(Objects::toString).orElse(null))); + headers.compute(MqttHeaders.CORRELATION_DATA, (k, v) -> + mapPropertyIfMatch(k, mqtt5Publish.getCorrelationData().orElse(null))); + + var userProperties = mqtt5Publish.getUserProperties().asList(); + for (Mqtt5UserProperty userProperty : userProperties) { + String name = userProperty.getName().toString(); + if (shouldMapHeader(name, this.inboundHeaderNames)) { + headers.put(name, userProperty.getValue().toString()); + } + } + return headers; + } + + private @Nullable Object mapPropertyIfMatch(String headerName, @Nullable Object value) { + return (value != null && shouldMapHeader(headerName, this.inboundHeaderNames)) ? value : null; + } + + private static boolean shouldMapHeader(String headerName, String[] patterns) { + for (String pattern : patterns) { + if (PatternMatchUtils.simpleMatch(pattern, headerName)) { + LOGGER.debug(LogMessage.format("headerName=[%s] WILL be mapped, matched pattern=%s", + headerName, pattern)); + return true; + } + } + LOGGER.debug(LogMessage.format("headerName=[%s] WILL NOT be mapped", headerName)); + return false; + } + +} diff --git a/spring-integration-hivemq/src/main/java/com/springframework/integration/hivemq/support/MqttHeaders.java b/spring-integration-hivemq/src/main/java/com/springframework/integration/hivemq/support/MqttHeaders.java new file mode 100644 index 00000000000..f1da53dd7d9 --- /dev/null +++ b/spring-integration-hivemq/src/main/java/com/springframework/integration/hivemq/support/MqttHeaders.java @@ -0,0 +1,57 @@ +/* + * 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.support; + +/** + * Spring Integration headers. + * + * @author Gary Russell + * + * @since 7.2 + */ +public final class MqttHeaders { + + public static final String PREFIX = "mqtt_"; + + public static final String QOS = PREFIX + "qos"; + + public static final String ID = PREFIX + "id"; + + public static final String RECEIVED_QOS = PREFIX + "receivedQos"; + + public static final String DUPLICATE = PREFIX + "duplicate"; + + public static final String RETAINED = PREFIX + "retained"; + + public static final String RECEIVED_RETAINED = PREFIX + "receivedRetained"; + + public static final String TOPIC = PREFIX + "topic"; + + public static final String RECEIVED_TOPIC = PREFIX + "receivedTopic"; + + public static final String MESSAGE_EXPIRY_INTERVAL = PREFIX + "messageExpiryInterval"; + + public static final String TOPIC_ALIAS = PREFIX + "topicAlias"; + + public static final String RESPONSE_TOPIC = PREFIX + "responseTopic"; + + public static final String CORRELATION_DATA = PREFIX + "correlationData"; + + private MqttHeaders() { + } + +} diff --git a/spring-integration-hivemq/src/main/java/com/springframework/integration/hivemq/support/package-info.java b/spring-integration-hivemq/src/main/java/com/springframework/integration/hivemq/support/package-info.java new file mode 100644 index 00000000000..81a25b85ada --- /dev/null +++ b/spring-integration-hivemq/src/main/java/com/springframework/integration/hivemq/support/package-info.java @@ -0,0 +1,5 @@ +/** + * Provides support classes for MqttAdapter components based on hivemq. + */ +@org.jspecify.annotations.NullMarked +package com.springframework.integration.hivemq.support; \ No newline at end of file diff --git a/spring-integration-hivemq/src/test/java/com/springframework/integration/hivemq/HiveMQContainer.java b/spring-integration-hivemq/src/test/java/com/springframework/integration/hivemq/HiveMQContainer.java new file mode 100644 index 00000000000..3274b32d2d5 --- /dev/null +++ b/spring-integration-hivemq/src/test/java/com/springframework/integration/hivemq/HiveMQContainer.java @@ -0,0 +1,56 @@ +/* + * 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 com.github.dockerjava.api.model.ExposedPort; +import com.github.dockerjava.api.model.HostConfig; +import com.github.dockerjava.api.model.PortBinding; +import com.github.dockerjava.api.model.Ports; +import org.junit.jupiter.api.BeforeAll; +import org.testcontainers.containers.GenericContainer; +import org.testcontainers.junit.jupiter.Testcontainers; + +/** + * The base contract for JUnit tests based on the container for HiveMQ MQTT broker. + * The Testcontainers 'reuse' option must be disabled,so, Ryuk container is started + * and will clean all the containers up from this test suite after JVM exit. + * Since the HiveMQ container instance is shared via static property, it is going to be + * started only once per JVM, therefore the target Docker container is reused automatically. + * + * @author Jiandong Ma + * + * @since 7.2 + */ +@Testcontainers(disabledWithoutDocker = true) +public interface HiveMQContainer { + + int CONTAINER_PORT = 1883; + + int MAPPED_PORT = 1883; + + GenericContainer HIVEMQ_CONTAINER = new GenericContainer<>("hivemq/hivemq-ce:2024.3") + .withExposedPorts(CONTAINER_PORT) + .withCreateContainerCmdModifier(cmd -> cmd.withHostConfig(new HostConfig() + .withPortBindings(new PortBinding(Ports.Binding.bindPort(MAPPED_PORT), new ExposedPort(CONTAINER_PORT))) + )); + + @BeforeAll + static void startContainer() { + HIVEMQ_CONTAINER.start(); + } + +} diff --git a/spring-integration-hivemq/src/test/java/com/springframework/integration/hivemq/inbound/Mqtt3MessageDrivenChannelAdapterTests.java b/spring-integration-hivemq/src/test/java/com/springframework/integration/hivemq/inbound/Mqtt3MessageDrivenChannelAdapterTests.java new file mode 100644 index 00000000000..0675a2f3a53 --- /dev/null +++ b/spring-integration-hivemq/src/test/java/com/springframework/integration/hivemq/inbound/Mqtt3MessageDrivenChannelAdapterTests.java @@ -0,0 +1,231 @@ +/* + * 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.inbound; + +import java.nio.charset.StandardCharsets; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executor; +import java.util.concurrent.TimeUnit; + +import com.hivemq.client.internal.mqtt.message.publish.mqtt3.Mqtt3PublishViewBuilder; +import com.hivemq.client.mqtt.datatypes.MqttQos; +import com.hivemq.client.mqtt.mqtt3.Mqtt3AsyncClient; +import com.hivemq.client.mqtt.mqtt3.Mqtt3Client; +import com.hivemq.client.mqtt.mqtt3.message.publish.Mqtt3Publish; +import com.springframework.integration.hivemq.HiveMQContainer; +import com.springframework.integration.hivemq.event.MqttSubscribedEvent; +import com.springframework.integration.hivemq.support.MqttHeaders; +import org.assertj.core.api.Assertions; +import org.assertj.core.api.InstanceOfAssertFactories; +import org.junit.jupiter.api.Test; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.event.EventListener; +import org.springframework.integration.IntegrationMessageHeaderAccessor; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.config.EnableIntegration; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHeaders; +import org.springframework.messaging.converter.AbstractMessageConverter; +import org.springframework.messaging.converter.SmartMessageConverter; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; + +/** + * @author Jiandong Ma + * + * @since 7.2 + */ +@SpringJUnitConfig +@DirtiesContext +class Mqtt3MessageDrivenChannelAdapterTests implements HiveMQContainer { + + static final String CAR_DEVICE_TOPIC_WILDCARD = "mqtt-v3-inbound-car-device/#"; + + static final String CAR_DEVICE_TOPIC_1 = "mqtt-v3-inbound-car-device/1"; + + static final String CAR_DEVICE_TOPIC_2 = "mqtt-v3-inbound-car-device/2"; + + static final CountDownLatch carDeviceWildcardTopicSubscribedLatch = new CountDownLatch(1); + + static final String PET_DEVICE_TOPIC = "mqtt-v3-inbound-pet-device"; + + static final CountDownLatch petDeviceTopicSubscribedLatch = new CountDownLatch(1); + + @Autowired + Mqtt3AsyncClient mqtt3AsyncClient; + + @Autowired + QueueChannel carDeviceOutputChannel; + + @Autowired + QueueChannel petDeviceOutputChannel; + + @Test + void testCarDeviceWildcardTopic() throws InterruptedException { + // Ensure subscription done first. + boolean subscribed = carDeviceWildcardTopicSubscribedLatch.await(10000, TimeUnit.MILLISECONDS); + Assertions.assertThat(subscribed).isTrue(); + // Given + mqtt3AsyncClient.publishWith().topic(CAR_DEVICE_TOPIC_1) + .payload("car-payload-1".getBytes()) + .qos(MqttQos.AT_LEAST_ONCE) + .send() + .orTimeout(10000, TimeUnit.MILLISECONDS) + .join(); + mqtt3AsyncClient.publishWith().topic(CAR_DEVICE_TOPIC_2) + .payload("car-payload-2".getBytes()) + .qos(MqttQos.AT_MOST_ONCE) + .send() + .orTimeout(10000, TimeUnit.MILLISECONDS) + .join(); + // When + Message firstCarMessage = carDeviceOutputChannel.receive(10000); + Message secondCarMessage = carDeviceOutputChannel.receive(10000); + // Then + Assertions.assertThat(firstCarMessage) + .isNotNull() + .returns("car-payload-1", Message::getPayload) + .extracting(Message::getHeaders) + .asInstanceOf(InstanceOfAssertFactories.type(MessageHeaders.class)) + .satisfies(headers -> Assertions.assertThat(headers) + .containsEntry(MqttHeaders.RECEIVED_QOS, MqttQos.AT_LEAST_ONCE) + .containsEntry(MqttHeaders.RECEIVED_TOPIC, CAR_DEVICE_TOPIC_1)); + Assertions.assertThat(secondCarMessage) + .isNotNull() + .returns("car-payload-2", Message::getPayload) + .extracting(Message::getHeaders) + .asInstanceOf(InstanceOfAssertFactories.type(MessageHeaders.class)) + .satisfies(headers -> Assertions.assertThat(headers) + .containsEntry(MqttHeaders.RECEIVED_QOS, MqttQos.AT_MOST_ONCE) // QoS Downgrade + .containsEntry(MqttHeaders.RECEIVED_TOPIC, CAR_DEVICE_TOPIC_2)); + } + + @Test + void testPetDeviceTopic() throws InterruptedException { + // Ensure subscription done first. + boolean subscribed = petDeviceTopicSubscribedLatch.await(10000, TimeUnit.MILLISECONDS); + Assertions.assertThat(subscribed).isTrue(); + // Given + Mqtt3Publish mqtt3Publish = new Mqtt3PublishViewBuilder.Default() + .topic(PET_DEVICE_TOPIC) + .qos(MqttQos.AT_LEAST_ONCE) + .payload("pet-payload-1".getBytes()) + .build(); + mqtt3AsyncClient.publish(mqtt3Publish) + .orTimeout(10000, TimeUnit.MILLISECONDS) + .join(); + // When + Message petDeviceMessage = petDeviceOutputChannel.receive(10000); + // Then + Assertions.assertThat(petDeviceMessage) + .isNotNull() + .returns("pet-payload-1".getBytes(), Message::getPayload) + .extracting(m -> m.getHeaders().get(IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK)) + .isNotNull() + .asInstanceOf(InstanceOfAssertFactories.type(Mqtt3Publish.class)) + .satisfies(mqttPublish -> { + Assertions.assertThat(mqttPublish).isEqualTo(mqtt3Publish); + mqttPublish.acknowledge(); // manual ack + }); + } + + @Configuration(proxyBeanMethods = false) + @EnableIntegration + static class Config { + + @Bean + Mqtt3AsyncClient mqtt3AsyncClient() { + return Mqtt3Client.builder() + .serverHost(HIVEMQ_CONTAINER.getHost()) + .serverPort(HIVEMQ_CONTAINER.getFirstMappedPort()) + .buildAsync(); + } + + @Bean + QueueChannel carDeviceOutputChannel() { + return new QueueChannel(); + } + + @Bean + Mqtt3MessageDrivenChannelAdapter carDeviceInboundChannelAdapter(Mqtt3AsyncClient mqtt3AsyncClient, + QueueChannel carDeviceOutputChannel, SmartMessageConverter mqttStringToBytesConverter) { + var adapter = new Mqtt3MessageDrivenChannelAdapter(mqtt3AsyncClient, CAR_DEVICE_TOPIC_WILDCARD); + adapter.setOutputChannel(carDeviceOutputChannel); + adapter.setQos(MqttQos.AT_LEAST_ONCE); + adapter.setPayloadType(String.class); + adapter.setMessageConverter(mqttStringToBytesConverter); + return adapter; + } + + @Bean + QueueChannel petDeviceOutputChannel() { + return new QueueChannel(); + } + + @Bean + Mqtt3MessageDrivenChannelAdapter petDeviceInboundChannelAdapter(Mqtt3AsyncClient mqtt3AsyncClient, + QueueChannel petDeviceOutputChannel, Executor executor) { + var adapter = new Mqtt3MessageDrivenChannelAdapter(mqtt3AsyncClient, PET_DEVICE_TOPIC); + adapter.setOutputChannel(petDeviceOutputChannel); + adapter.setManualAcknowledgement(true); + adapter.setExecutor(executor); + return adapter; + } + + @EventListener + void mqttEvents(MqttSubscribedEvent event) { + String beanName = ((Mqtt3MessageDrivenChannelAdapter) event.getSource()).getBeanName(); + if (beanName.equals("carDeviceInboundChannelAdapter")) { + carDeviceWildcardTopicSubscribedLatch.countDown(); + } + else if (beanName.equals("petDeviceInboundChannelAdapter")) { + petDeviceTopicSubscribedLatch.countDown(); + } + } + + @Bean + public SmartMessageConverter mqttStringToBytesConverter() { + return new AbstractMessageConverter() { + + @Override + protected boolean supports(Class clazz) { + return true; + } + + @Override + protected Object convertFromInternal(Message message, Class targetClass, + Object conversionHint) { + + return message.getPayload().toString().getBytes(StandardCharsets.UTF_8); + } + + @Override + protected Object convertToInternal(Object payload, MessageHeaders headers, + Object conversionHint) { + + return new String((byte[]) payload); + } + + }; + } + + } + +} diff --git a/spring-integration-hivemq/src/test/java/com/springframework/integration/hivemq/inbound/Mqtt3ResubscribeAfterAutomaticReconnectTests.java b/spring-integration-hivemq/src/test/java/com/springframework/integration/hivemq/inbound/Mqtt3ResubscribeAfterAutomaticReconnectTests.java new file mode 100644 index 00000000000..4c6737b7a81 --- /dev/null +++ b/spring-integration-hivemq/src/test/java/com/springframework/integration/hivemq/inbound/Mqtt3ResubscribeAfterAutomaticReconnectTests.java @@ -0,0 +1,125 @@ +/* + * 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.inbound; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import com.hivemq.client.internal.mqtt.message.connect.mqtt3.Mqtt3ConnectViewBuilder; +import com.hivemq.client.mqtt.datatypes.MqttQos; +import com.hivemq.client.mqtt.mqtt3.Mqtt3AsyncClient; +import com.hivemq.client.mqtt.mqtt3.Mqtt3Client; +import com.springframework.integration.hivemq.HiveMQContainer; +import com.springframework.integration.hivemq.event.MqttSubscribedEvent; +import org.assertj.core.api.Assertions; +import org.junit.jupiter.api.Test; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.event.EventListener; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.config.EnableIntegration; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * @author Jiandong Ma + * + * @since 7.2 + */ +@SpringJUnitConfig +@DirtiesContext +class Mqtt3ResubscribeAfterAutomaticReconnectTests implements HiveMQContainer { + + static final String TOPIC = "topic-for-mqtt-v3-automatic-reconnect"; + + static final CountDownLatch subscribeFirstLatch = new CountDownLatch(1); + + static final CountDownLatch connectSecondLatch = new CountDownLatch(2); + + @Autowired + Mqtt3AsyncClient mqtt3AsyncClient; + + @Autowired + QueueChannel outputChannel; + + @Test + void messageReceivedAfterAutomaticReConnection() throws InterruptedException { + // subscribe done + assertThat(subscribeFirstLatch.await(10, TimeUnit.SECONDS)).isTrue(); + // Given + mqtt3AsyncClient.publishWith().topic(TOPIC).payload("payload-1".getBytes()) + .send().orTimeout(10000, TimeUnit.MILLISECONDS).join(); + // Then + Assertions.assertThat(outputChannel.receive(10000)).isNotNull(); + // broken down and up + HIVEMQ_CONTAINER.stop(); + HIVEMQ_CONTAINER.start(); + // await reconnect, manual resubscribe not need. + Assertions.assertThat(connectSecondLatch.await(20, TimeUnit.SECONDS)).isTrue(); + // Given + mqtt3AsyncClient.publishWith().topic(TOPIC).payload("payload-2".getBytes()) + .send().orTimeout(10000, TimeUnit.MILLISECONDS).join(); + // Then + Assertions.assertThat(outputChannel.receive(10000)).isNotNull(); + } + + @Configuration(proxyBeanMethods = false) + @EnableIntegration + static class Config { + + @Bean + Mqtt3AsyncClient mqtt3AsyncClient() { + return Mqtt3Client.builder() + .serverHost(HIVEMQ_CONTAINER.getHost()) + .serverPort(HIVEMQ_CONTAINER.getFirstMappedPort()) + .automaticReconnect() + .initialDelay(1, TimeUnit.SECONDS) + .maxDelay(2, TimeUnit.SECONDS) + .applyAutomaticReconnect() + .addConnectedListener(ctx -> connectSecondLatch.countDown()) + .buildAsync(); + } + + @Bean + QueueChannel outputChannel() { + return new QueueChannel(); + } + + @Bean + Mqtt3MessageDrivenChannelAdapter mqtt5InboundChannelAdapter(Mqtt3AsyncClient mqtt3AsyncClient, + QueueChannel outputChannel) { + var adapter = new Mqtt3MessageDrivenChannelAdapter(mqtt3AsyncClient, TOPIC); + adapter.setOutputChannel(outputChannel); + adapter.setQos(MqttQos.AT_LEAST_ONCE); + adapter.setMqtt3ConnectView(new Mqtt3ConnectViewBuilder.Default() + .cleanSession(true) // looks even cleanStart is true, resubscribe can automatic happens after reconnect. + .build()); + return adapter; + } + + @EventListener + void mqttEvents(MqttSubscribedEvent event) { + subscribeFirstLatch.countDown(); + } + + } + +} diff --git a/spring-integration-hivemq/src/test/java/com/springframework/integration/hivemq/inbound/Mqtt5MessageDrivenChannelAdapterTests.java b/spring-integration-hivemq/src/test/java/com/springframework/integration/hivemq/inbound/Mqtt5MessageDrivenChannelAdapterTests.java new file mode 100644 index 00000000000..bb6085509c0 --- /dev/null +++ b/spring-integration-hivemq/src/test/java/com/springframework/integration/hivemq/inbound/Mqtt5MessageDrivenChannelAdapterTests.java @@ -0,0 +1,241 @@ +/* + * 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.inbound; + +import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executor; +import java.util.concurrent.TimeUnit; + +import com.hivemq.client.internal.mqtt.datatypes.MqttUserPropertyImpl; +import com.hivemq.client.internal.mqtt.message.publish.MqttPublishBuilder; +import com.hivemq.client.mqtt.MqttClient; +import com.hivemq.client.mqtt.datatypes.MqttQos; +import com.hivemq.client.mqtt.mqtt5.Mqtt5AsyncClient; +import com.hivemq.client.mqtt.mqtt5.datatypes.Mqtt5UserProperties; +import com.hivemq.client.mqtt.mqtt5.message.publish.Mqtt5Publish; +import com.springframework.integration.hivemq.HiveMQContainer; +import com.springframework.integration.hivemq.event.MqttSubscribedEvent; +import com.springframework.integration.hivemq.support.MqttHeaders; +import org.assertj.core.api.Assertions; +import org.assertj.core.api.InstanceOfAssertFactories; +import org.junit.jupiter.api.Test; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.event.EventListener; +import org.springframework.integration.IntegrationMessageHeaderAccessor; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.config.EnableIntegration; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHeaders; +import org.springframework.messaging.converter.AbstractMessageConverter; +import org.springframework.messaging.converter.SmartMessageConverter; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; + +/** + * @author Jiandong Ma + * + * @since 7.2 + */ +@SpringJUnitConfig +@DirtiesContext +class Mqtt5MessageDrivenChannelAdapterTests implements HiveMQContainer { + + static final String CAR_DEVICE_TOPIC_WILDCARD = "mqtt-v5-inbound-car-device/#"; + + static final String CAR_DEVICE_TOPIC_1 = "mqtt-v5-inbound-car-device/1"; + + static final String CAR_DEVICE_TOPIC_2 = "mqtt-v5-inbound-car-device/2"; + + static final CountDownLatch carDeviceWildcardTopicSubscribedLatch = new CountDownLatch(1); + + static final String PET_DEVICE_TOPIC = "mqtt-v5-inbound-pet-device"; + + static final CountDownLatch petDeviceTopicSubscribedLatch = new CountDownLatch(1); + + @Autowired + Mqtt5AsyncClient mqtt5AsyncClient; + + @Autowired + QueueChannel carDeviceOutputChannel; + + @Autowired + QueueChannel petDeviceOutputChannel; + + @Test + void testCarDeviceWildcardTopic() throws InterruptedException { + // Ensure subscription done first. + boolean subscribed = carDeviceWildcardTopicSubscribedLatch.await(10000, TimeUnit.MILLISECONDS); + Assertions.assertThat(subscribed).isTrue(); + // Given + mqtt5AsyncClient.publishWith().topic(CAR_DEVICE_TOPIC_1) + .payload("car-payload-1".getBytes()) + .qos(MqttQos.AT_LEAST_ONCE) + .contentType("text/plain") + .correlationData("correlation-id".getBytes()) + .userProperties(Mqtt5UserProperties.of(MqttUserPropertyImpl.of("custom-prop", "custom-val"))) + .send() + .orTimeout(10000, TimeUnit.MILLISECONDS) + .join(); + mqtt5AsyncClient.publishWith().topic(CAR_DEVICE_TOPIC_2) + .payload("car-payload-2".getBytes()) + .qos(MqttQos.AT_MOST_ONCE) + .send() + .orTimeout(10000, TimeUnit.MILLISECONDS) + .join(); + // When + Message firstCarMessage = carDeviceOutputChannel.receive(10000); + Message secondCarMessage = carDeviceOutputChannel.receive(10000); + // Then + Assertions.assertThat(firstCarMessage) + .isNotNull() + .returns("car-payload-1", Message::getPayload) + .extracting(Message::getHeaders) + .asInstanceOf(InstanceOfAssertFactories.type(MessageHeaders.class)) + .satisfies(headers -> Assertions.assertThat(headers) + .containsEntry(MqttHeaders.RECEIVED_QOS, MqttQos.AT_LEAST_ONCE) + .containsEntry(MqttHeaders.RECEIVED_TOPIC, CAR_DEVICE_TOPIC_1) + .containsEntry(MessageHeaders.CONTENT_TYPE, "text/plain") + .containsEntry(MqttHeaders.CORRELATION_DATA, ByteBuffer.wrap("correlation-id".getBytes())) + .containsEntry("custom-prop", "custom-val")); + Assertions.assertThat(secondCarMessage) + .isNotNull() + .returns("car-payload-2", Message::getPayload) + .extracting(Message::getHeaders) + .asInstanceOf(InstanceOfAssertFactories.type(MessageHeaders.class)) + .satisfies(headers -> Assertions.assertThat(headers) + .containsEntry(MqttHeaders.RECEIVED_QOS, MqttQos.AT_MOST_ONCE) // QoS Downgrade + .containsEntry(MqttHeaders.RECEIVED_TOPIC, CAR_DEVICE_TOPIC_2)); + } + + @Test + void testPetDeviceTopic() throws InterruptedException { + // Ensure subscription done first. + boolean subscribed = petDeviceTopicSubscribedLatch.await(10000, TimeUnit.MILLISECONDS); + Assertions.assertThat(subscribed).isTrue(); + // Given + Mqtt5Publish mqtt5Publish = new MqttPublishBuilder.Default() + .topic(PET_DEVICE_TOPIC) + .qos(MqttQos.AT_LEAST_ONCE) + .payload("pet-payload-1".getBytes()) + .build(); + mqtt5AsyncClient.publish(mqtt5Publish) + .orTimeout(10000, TimeUnit.MILLISECONDS) + .join(); + // When + Message petDeviceMessage = petDeviceOutputChannel.receive(10000); + // Then + Assertions.assertThat(petDeviceMessage) + .isNotNull() + .returns("pet-payload-1".getBytes(), Message::getPayload) + .extracting(m -> m.getHeaders().get(IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK)) + .isNotNull() + .asInstanceOf(InstanceOfAssertFactories.type(Mqtt5Publish.class)) + .satisfies(mqttPublish -> { + Assertions.assertThat(mqttPublish).isEqualTo(mqtt5Publish); + mqttPublish.acknowledge(); // manual ack + }); + } + + @Configuration(proxyBeanMethods = false) + @EnableIntegration + static class Config { + + @Bean + Mqtt5AsyncClient mqtt5AsyncClient() { + return MqttClient.builder() + .useMqttVersion5() + .serverHost(HIVEMQ_CONTAINER.getHost()) + .serverPort(HIVEMQ_CONTAINER.getFirstMappedPort()) + .buildAsync(); + } + + @Bean + QueueChannel carDeviceOutputChannel() { + return new QueueChannel(); + } + + @Bean + Mqtt5MessageDrivenChannelAdapter carDeviceInboundChannelAdapter(Mqtt5AsyncClient mqtt5AsyncClient, + QueueChannel carDeviceOutputChannel, SmartMessageConverter mqttStringToBytesConverter) { + var adapter = new Mqtt5MessageDrivenChannelAdapter(mqtt5AsyncClient, CAR_DEVICE_TOPIC_WILDCARD); + adapter.setOutputChannel(carDeviceOutputChannel); + adapter.setQos(MqttQos.AT_LEAST_ONCE); + adapter.setPayloadType(String.class); + adapter.setMessageConverter(mqttStringToBytesConverter); + return adapter; + } + + @Bean + QueueChannel petDeviceOutputChannel() { + return new QueueChannel(); + } + + @Bean + Mqtt5MessageDrivenChannelAdapter petDeviceInboundChannelAdapter(Mqtt5AsyncClient mqtt5AsyncClient, + QueueChannel petDeviceOutputChannel, Executor executor) { + var adapter = new Mqtt5MessageDrivenChannelAdapter(mqtt5AsyncClient, PET_DEVICE_TOPIC); + adapter.setOutputChannel(petDeviceOutputChannel); + adapter.setManualAcknowledgement(true); + adapter.setExecutor(executor); + return adapter; + } + + @EventListener + void mqttEvents(MqttSubscribedEvent event) { + String beanName = ((Mqtt5MessageDrivenChannelAdapter) event.getSource()).getBeanName(); + if (beanName.equals("carDeviceInboundChannelAdapter")) { + carDeviceWildcardTopicSubscribedLatch.countDown(); + } + else if (beanName.equals("petDeviceInboundChannelAdapter")) { + petDeviceTopicSubscribedLatch.countDown(); + } + } + + @Bean + public SmartMessageConverter mqttStringToBytesConverter() { + return new AbstractMessageConverter() { + + @Override + protected boolean supports(Class clazz) { + return true; + } + + @Override + protected Object convertFromInternal(Message message, Class targetClass, + Object conversionHint) { + + return message.getPayload().toString().getBytes(StandardCharsets.UTF_8); + } + + @Override + protected Object convertToInternal(Object payload, MessageHeaders headers, + Object conversionHint) { + + return new String((byte[]) payload); + } + + }; + } + + } + +} diff --git a/spring-integration-hivemq/src/test/java/com/springframework/integration/hivemq/inbound/Mqtt5ResubscribeAfterAutomaticReconnectTests.java b/spring-integration-hivemq/src/test/java/com/springframework/integration/hivemq/inbound/Mqtt5ResubscribeAfterAutomaticReconnectTests.java new file mode 100644 index 00000000000..19df5070a16 --- /dev/null +++ b/spring-integration-hivemq/src/test/java/com/springframework/integration/hivemq/inbound/Mqtt5ResubscribeAfterAutomaticReconnectTests.java @@ -0,0 +1,134 @@ +/* + * 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.inbound; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import com.hivemq.client.internal.mqtt.message.connect.MqttConnectBuilder; +import com.hivemq.client.internal.mqtt.message.disconnect.MqttDisconnect; +import com.hivemq.client.internal.mqtt.message.subscribe.MqttSubscription; +import com.hivemq.client.mqtt.datatypes.MqttQos; +import com.hivemq.client.mqtt.mqtt5.Mqtt5AsyncClient; +import com.hivemq.client.mqtt.mqtt5.Mqtt5Client; +import com.springframework.integration.hivemq.HiveMQContainer; +import com.springframework.integration.hivemq.event.MqttSubscribedEvent; +import com.springframework.integration.hivemq.support.Mqtt5HeaderMapper; +import org.assertj.core.api.Assertions; +import org.junit.jupiter.api.Test; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.event.EventListener; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.config.EnableIntegration; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * @author Jiandong Ma + * + * @since 7.2 + */ +@SpringJUnitConfig +@DirtiesContext +class Mqtt5ResubscribeAfterAutomaticReconnectTests implements HiveMQContainer { + + static final String TOPIC = "topic-for-mqtt-v5-automatic-reconnect"; + + static final CountDownLatch subscribeFirstLatch = new CountDownLatch(1); + + static final CountDownLatch connectSecondLatch = new CountDownLatch(2); + + @Autowired + Mqtt5AsyncClient mqtt5AsyncClient; + + @Autowired + QueueChannel outputChannel; + + @Test + void messageReceivedAfterAutomaticReConnection() throws InterruptedException { + // subscribe done + assertThat(subscribeFirstLatch.await(10, TimeUnit.SECONDS)).isTrue(); + // Given + mqtt5AsyncClient.publishWith().topic(TOPIC).payload("payload-1".getBytes()) + .send().orTimeout(10000, TimeUnit.MILLISECONDS).join(); + // Then + Assertions.assertThat(outputChannel.receive(10000)).isNotNull(); + // broken down and up + HIVEMQ_CONTAINER.stop(); + HIVEMQ_CONTAINER.start(); + // await reconnect, manual resubscribe not need. + Assertions.assertThat(connectSecondLatch.await(20, TimeUnit.SECONDS)).isTrue(); + // Given + mqtt5AsyncClient.publishWith().topic(TOPIC).payload("payload-2".getBytes()) + .send().orTimeout(10000, TimeUnit.MILLISECONDS).join(); + // Then + Assertions.assertThat(outputChannel.receive(10000)).isNotNull(); + } + + @Configuration(proxyBeanMethods = false) + @EnableIntegration + static class Config { + + @Bean + Mqtt5AsyncClient mqtt5AsyncClient() { + return Mqtt5Client.builder() + .serverHost(HIVEMQ_CONTAINER.getHost()) + .serverPort(HIVEMQ_CONTAINER.getFirstMappedPort()) + .automaticReconnect() + .initialDelay(1, TimeUnit.SECONDS) + .maxDelay(2, TimeUnit.SECONDS) + .applyAutomaticReconnect() + .addConnectedListener(ctx -> connectSecondLatch.countDown()) + .buildAsync(); + } + + @Bean + QueueChannel outputChannel() { + return new QueueChannel(); + } + + @Bean + Mqtt5MessageDrivenChannelAdapter mqtt5InboundChannelAdapter(Mqtt5AsyncClient mqtt5AsyncClient, + QueueChannel outputChannel) { + var adapter = new Mqtt5MessageDrivenChannelAdapter(mqtt5AsyncClient, TOPIC); + adapter.setOutputChannel(outputChannel); + adapter.setQos(MqttQos.AT_LEAST_ONCE); + adapter.setMqttConnect(new MqttConnectBuilder.Default() + .cleanStart(true) // looks even cleanStart is true, resubscribe can automatic happens after reconnect. + .build()); + // below are default, for line coverage only + adapter.setHeaderMapper(new Mqtt5HeaderMapper()); + adapter.setNoLocal(MqttSubscription.DEFAULT_NO_LOCAL); + adapter.setRetainHandling(MqttSubscription.DEFAULT_RETAIN_HANDLING); + adapter.setRetainAsPublished(MqttSubscription.DEFAULT_RETAIN_AS_PUBLISHED); + adapter.setMqttDisconnect(MqttDisconnect.DEFAULT); + return adapter; + } + + @EventListener + void mqttEvents(MqttSubscribedEvent event) { + subscribeFirstLatch.countDown(); + } + + } + +}