Skip to content

Commit a7c81c3

Browse files
mjd507mjd507
committed
Add Inbound Channel Adapter for MQTT - based on hivemq
Related to: #3102 Signed-off-by: Jiandong Ma <jiandong.ma.cn@gmail.com>
1 parent fe5e7d8 commit a7c81c3

25 files changed

Lines changed: 2257 additions & 0 deletions

‎gradle/libs.versions.toml‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,7 @@ h2Version = "2.4.240"
3636
hamcrestVersion = "3.0"
3737
hazelcastVersion = "5.7.0"
3838
hibernateVersion = "7.4.5.Final"
39+
hivemqMqttClientVersion="1.3.17"
3940
hsqldbVersion = "2.7.4"
4041
jackson3Version = "3.2.1"
4142
jacksonVersion = "2.22.1"
@@ -93,6 +94,7 @@ com-github-spotbugs-annotations = { module = 'com.github.spotbugs:spotbugs-annot
9394
com-google-protobuf-bom = { module = "com.google.protobuf:protobuf-bom", version.ref = "protobufVersion" }
9495
com-h2database-h2 = { module = "com.h2database:h2", version.ref = "h2Version" }
9596
com-hazelcast = { module = "com.hazelcast:hazelcast", version.ref = "hazelcastVersion" }
97+
com-hivemq-mqtt-client = { module="com.hivemq:hivemq-mqtt-client", version.ref = "hivemqMqttClientVersion"}
9698
com-icegreen-greenmail = { module = "com.icegreen:greenmail", version.ref = "greenmailVersion" }
9799
com-jayway-jsonpath = { module = "com.jayway.jsonpath:json-path", version.ref = "jsonpathVersion" }
98100
com-mysql-connector = { module = "com.mysql:mysql-connector-j", version.ref = "mysqlVersion" }

‎spring-integration-mqtt-client/build.gradle‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,4 @@
1+
description = 'Spring Integration MQTT Support'
2+
dependencies {
3+
api libs.com.hivemq.mqtt.client
4+
}

‎spring-integration-mqtt-client/src/main/java/com/springframework/integration/mqtt/client/core/AbstractMqttClientManager.java‎

Lines changed: 151 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,151 @@
1+
/*
2+
* Copyright 2026-present the original author or authors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* https://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package com.springframework.integration.mqtt.client.core;
18+
19+
import java.util.Collections;
20+
import java.util.HashSet;
21+
import java.util.Set;
22+
import java.util.concurrent.locks.Lock;
23+
import java.util.concurrent.locks.ReentrantLock;
24+
25+
import com.hivemq.client.mqtt.MqttClient;
26+
import com.springframework.integration.mqtt.client.inbound.AbstractMqttMessageDrivenChannelAdapter;
27+
import org.apache.commons.logging.Log;
28+
import org.apache.commons.logging.LogFactory;
29+
30+
import org.springframework.context.ApplicationEventPublisher;
31+
import org.springframework.context.ApplicationEventPublisherAware;
32+
import org.springframework.context.SmartLifecycle;
33+
import org.springframework.util.Assert;
34+
35+
/**
36+
* Abstract class for MQTT client managers which can be a base for any common v3/v5 client manager implementation.
37+
* Contains some basic utility and implementation-agnostic fields and methods.
38+
*
39+
* @param <T> MQTT client type
40+
* @param <B> MQTT client builder
41+
*
42+
* @author Jiandong Ma
43+
*
44+
* @since 7.2
45+
*/
46+
public abstract class AbstractMqttClientManager<T extends MqttClient, B>
47+
implements MqttClientManager<T>, ApplicationEventPublisherAware {
48+
49+
protected final Log logger = LogFactory.getLog(this.getClass());
50+
51+
protected final Lock lock = new ReentrantLock();
52+
53+
private int phase = DEFAULT_MANAGER_PHASE;
54+
55+
protected long connectCompletionTimeout = CONNECT_COMPLETION_TIMEOUT;
56+
57+
protected long disconnectCompletionTimeout = DISCONNECT_COMPLETION_TIMEOUT;
58+
59+
protected final Set<ConnectCallback> connectCallbacks = Collections.synchronizedSet(new HashSet<>());
60+
61+
protected final B mqttClientBuilder;
62+
63+
@SuppressWarnings("NullAway.Init")
64+
protected T mqttClient;
65+
66+
@SuppressWarnings("NullAway.Init")
67+
protected ApplicationEventPublisher applicationEventPublisher;
68+
69+
protected AbstractMqttClientManager(B mqttClientBuilder) {
70+
this.mqttClientBuilder = mqttClientBuilder;
71+
}
72+
73+
@Override
74+
public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
75+
Assert.notNull(applicationEventPublisher, "'applicationEventPublisher' cannot be null");
76+
this.applicationEventPublisher = applicationEventPublisher;
77+
}
78+
79+
/**
80+
* Set the phase of component autostart in {@link SmartLifecycle}.
81+
* If the custom one is required, note that for the correct behavior it should be less than phase of
82+
* {@link AbstractMqttMessageDrivenChannelAdapter} implementations.
83+
* @see #getPhase
84+
*/
85+
public void setPhase(int phase) {
86+
this.phase = phase;
87+
}
88+
89+
/**
90+
* Set the completion timeout when connecting.
91+
* Default {@value #CONNECT_COMPLETION_TIMEOUT} milliseconds.
92+
* @param connectCompletionTimeout The timeout.
93+
*/
94+
public void setConnectCompletionTimeout(long connectCompletionTimeout) {
95+
this.connectCompletionTimeout = connectCompletionTimeout;
96+
}
97+
98+
/**
99+
* Set the completion timeout when disconnecting.
100+
* Default {@value #DISCONNECT_COMPLETION_TIMEOUT} milliseconds.
101+
* @param completionTimeout The timeout.
102+
*/
103+
public void setDisconnectCompletionTimeout(long completionTimeout) {
104+
this.disconnectCompletionTimeout = completionTimeout;
105+
}
106+
107+
@Override
108+
public void addCallback(ConnectCallback connectCallback) {
109+
this.connectCallbacks.add(connectCallback);
110+
}
111+
112+
@Override
113+
public boolean removeCallback(ConnectCallback connectCallback) {
114+
return this.connectCallbacks.remove(connectCallback);
115+
}
116+
117+
@Override
118+
public T getClient() {
119+
return this.mqttClient;
120+
}
121+
122+
/**
123+
* The phase of component auto-start in {@link SmartLifecycle}.
124+
* If the custom one is required, note that for the correct behavior it should be less than phase of
125+
* {@link AbstractMqttMessageDrivenChannelAdapter} implementations.
126+
* The default phase is {@link #DEFAULT_MANAGER_PHASE}.
127+
* @return {@link SmartLifecycle} autostart phase
128+
* @see #setPhase
129+
*/
130+
@Override
131+
public int getPhase() {
132+
return this.phase;
133+
}
134+
135+
@Override
136+
public boolean isRunning() {
137+
return true;
138+
}
139+
140+
@Override
141+
public boolean isConnected() {
142+
this.lock.lock();
143+
try {
144+
return this.mqttClient.getState().isConnected();
145+
}
146+
finally {
147+
this.lock.unlock();
148+
}
149+
}
150+
151+
}

‎spring-integration-mqtt-client/src/main/java/com/springframework/integration/mqtt/client/core/Mqtt3ClientManager.java‎

Lines changed: 107 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,107 @@
1+
/*
2+
* Copyright 2026-present the original author or authors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* https://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
package com.springframework.integration.mqtt.client.core;
18+
19+
import java.util.concurrent.ExecutionException;
20+
import java.util.concurrent.TimeUnit;
21+
import java.util.concurrent.TimeoutException;
22+
23+
import com.hivemq.client.internal.mqtt.message.connect.mqtt3.Mqtt3ConnectView;
24+
import com.hivemq.client.mqtt.lifecycle.MqttClientConnectedContext;
25+
import com.hivemq.client.mqtt.lifecycle.MqttClientConnectedListener;
26+
import com.hivemq.client.mqtt.mqtt3.Mqtt3AsyncClient;
27+
import com.hivemq.client.mqtt.mqtt3.Mqtt3ClientBuilder;
28+
import com.hivemq.client.mqtt.mqtt3.message.connect.Mqtt3Connect;
29+
import com.springframework.integration.mqtt.client.event.MqttConnectionFailedEvent;
30+
import com.springframework.integration.mqtt.client.support.MqttClientBuilderHelper;
31+
32+
import org.springframework.util.Assert;
33+
34+
/**
35+
* A client manager implementation for MQTT v3 protocol.
36+
*
37+
* @author Jiandong Ma
38+
*
39+
* @since 7.2
40+
*/
41+
public class Mqtt3ClientManager extends AbstractMqttClientManager<Mqtt3AsyncClient, Mqtt3ClientBuilder>
42+
implements MqttClientConnectedListener {
43+
44+
private Mqtt3Connect mqttConnect = Mqtt3ConnectView.DEFAULT;
45+
46+
protected Mqtt3ClientManager(Mqtt3ClientBuilder mqttClientBuilder) {
47+
super(mqttClientBuilder);
48+
49+
this.mqttClient = MqttClientBuilderHelper.clone(mqttClientBuilder)
50+
.addConnectedListener(Mqtt3ClientManager.this)
51+
.buildAsync();
52+
53+
if (this.mqttClient.getConfig().getAutomaticReconnect().isEmpty()) {
54+
logger.info("If this `ClientManager` is used from message-driven channel adapters, " +
55+
"it is recommended to enable 'automaticReconnect' when set the 'mqttClientBuilder'. " +
56+
"Otherwise connection check and reconnect should be done manually.");
57+
}
58+
}
59+
60+
/**
61+
* Set the Connect message.
62+
* @param mqttConnect the mqttConnect
63+
*/
64+
public void setMqttConnect(Mqtt3Connect mqttConnect) {
65+
Assert.notNull(mqttConnect, "'mqttConnect' must not be null.");
66+
this.mqttConnect = mqttConnect;
67+
}
68+
69+
@Override
70+
public void start() {
71+
this.lock.lock();
72+
try {
73+
try {
74+
this.mqttClient.connect(this.mqttConnect).get(this.connectCompletionTimeout, TimeUnit.MILLISECONDS);
75+
}
76+
catch (InterruptedException | ExecutionException | TimeoutException ex) {
77+
applicationEventPublisher.publishEvent(new MqttConnectionFailedEvent(this, ex));
78+
logger.error("Could not start client manager", ex);
79+
}
80+
}
81+
finally {
82+
this.lock.unlock();
83+
}
84+
}
85+
86+
@Override
87+
public void stop() {
88+
this.lock.lock();
89+
try {
90+
try {
91+
this.mqttClient.disconnect().get(this.disconnectCompletionTimeout, TimeUnit.MILLISECONDS);
92+
}
93+
catch (ExecutionException | InterruptedException | TimeoutException e) {
94+
logger.error("Could not disconnect from the client", e);
95+
}
96+
}
97+
finally {
98+
this.lock.unlock();
99+
}
100+
}
101+
102+
@Override
103+
public void onConnected(MqttClientConnectedContext context) {
104+
connectCallbacks.forEach(connectCallback -> connectCallback.onClientConnected(context));
105+
}
106+
107+
}

0 commit comments

Comments
 (0)