Skip to content

Commit 285e0af

Browse files
BenCodezBenCodez
committed
Recover MQTT connections and subscriptions after broker restarts
1 parent f379ad6 commit 285e0af

5 files changed

Lines changed: 494 additions & 101 deletions

File tree

‎SimpleAPI/src/main/java/com/bencodez/simpleapi/servercomm/mqtt/MqttHandler.java‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -67,8 +67,7 @@ public void connect() throws Exception {
6767
}
6868

6969
public void disconnect() throws Exception {
70-
mqtt.disconnect();
71-
scheduler.shutdownNow();
70+
try { mqtt.close(); } finally { scheduler.shutdownNow(); }
7271
}
7372

7473
public boolean isConnected() {

‎SimpleAPI/src/main/java/com/bencodez/simpleapi/servercomm/mqtt/MqttServerComm.java‎

Lines changed: 248 additions & 99 deletions
Original file line numberDiff line numberDiff line change
@@ -1,114 +1,263 @@
11
package com.bencodez.simpleapi.servercomm.mqtt;
22

33
import java.nio.charset.StandardCharsets;
4+
import java.util.LinkedHashMap;
5+
import java.util.Objects;
6+
import java.util.concurrent.ScheduledFuture;
7+
import java.util.concurrent.ScheduledThreadPoolExecutor;
8+
import java.util.concurrent.TimeUnit;
49

5-
import org.eclipse.paho.client.mqttv3.IMqttActionListener;
610
import org.eclipse.paho.client.mqttv3.IMqttAsyncClient;
11+
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
712
import org.eclipse.paho.client.mqttv3.IMqttToken;
813
import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
14+
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
915
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
1016
import org.eclipse.paho.client.mqttv3.MqttException;
1117
import org.eclipse.paho.client.mqttv3.MqttMessage;
1218

1319
import com.bencodez.simpleapi.servercomm.codec.JsonEnvelope;
1420
import com.bencodez.simpleapi.servercomm.codec.JsonEnvelopeCodec;
1521

16-
/**
17-
* Very basic MQTT client wrapper using Eclipse Paho.
18-
* Provides connect/disconnect, publish, subscribe, and serverId access.
19-
* Supports username/password authentication.
20-
* Envelope helpers included.
21-
*/
22-
public class MqttServerComm {
23-
24-
private final String serverId;
25-
private final IMqttAsyncClient client;
26-
private final MqttConnectOptions connectOptions;
27-
28-
public MqttServerComm(String serverId, String brokerUrl, MqttConnectOptions opts) throws MqttException {
29-
this.serverId = serverId;
30-
this.client = new MqttAsyncClient(brokerUrl, serverId);
31-
this.connectOptions = opts;
32-
this.client.connect(connectOptions).waitForCompletion();
33-
}
34-
35-
public MqttServerComm(String serverId, String brokerUrl, String username, String password) throws MqttException {
36-
this.serverId = serverId;
37-
this.client = new MqttAsyncClient(brokerUrl, serverId);
38-
this.connectOptions = new MqttConnectOptions();
39-
this.connectOptions.setCleanSession(true);
40-
41-
if (username != null && !username.isEmpty()) {
42-
this.connectOptions.setUserName(username);
43-
}
44-
if (password != null && !password.isEmpty()) {
45-
this.connectOptions.setPassword(password.toCharArray());
46-
}
47-
48-
this.client.connect(connectOptions).waitForCompletion();
49-
}
50-
51-
public String getServerId() {
52-
return serverId;
53-
}
54-
55-
public void connect() throws MqttException {
56-
if (!client.isConnected()) {
57-
client.connect(connectOptions).waitForCompletion();
58-
}
59-
}
60-
61-
public void disconnect() throws MqttException {
62-
if (client.isConnected()) {
63-
client.disconnect().waitForCompletion();
64-
}
65-
}
66-
67-
public boolean isConnected() {
68-
return client.isConnected();
69-
}
70-
71-
public void publish(String topic, String payload, int qos, boolean retained) throws MqttException {
72-
MqttMessage msg = new MqttMessage(payload.getBytes(StandardCharsets.UTF_8));
73-
msg.setQos(qos);
74-
msg.setRetained(retained);
75-
client.publish(topic, msg);
76-
}
77-
78-
public void publishEnvelope(String topic, JsonEnvelope envelope, int qos, boolean retained) throws MqttException {
79-
publish(topic, JsonEnvelopeCodec.encode(envelope), qos, retained);
80-
}
81-
82-
public void subscribe(String topicFilter, int qos, MessageListener listener) throws MqttException {
83-
client.subscribe(topicFilter, qos, null, new IMqttActionListener() {
84-
@Override
85-
public void onSuccess(IMqttToken asyncActionToken) {
86-
}
87-
88-
@Override
89-
public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
90-
exception.printStackTrace();
91-
}
92-
}, (topic, msg) -> listener.messageArrived(topic, msg));
93-
}
94-
95-
public void subscribeEnvelopes(String topicFilter, int qos, EnvelopeListener listener) throws MqttException {
96-
subscribe(topicFilter, qos, (topic, msg) -> {
97-
String payload = new String(msg.getPayload(), StandardCharsets.UTF_8);
98-
JsonEnvelope env = JsonEnvelopeCodec.decode(payload);
99-
listener.envelopeArrived(topic, env);
100-
});
101-
}
102-
103-
public void unsubscribe(String topicFilter) throws MqttException {
104-
client.unsubscribe(topicFilter);
105-
}
106-
107-
public interface MessageListener {
108-
void messageArrived(String topic, MqttMessage message) throws Exception;
109-
}
110-
111-
public interface EnvelopeListener {
112-
void envelopeArrived(String topic, JsonEnvelope envelope) throws Exception;
113-
}
22+
/** MQTT communication with owned, bounded reconnect and subscription recovery. */
23+
public class MqttServerComm implements AutoCloseable {
24+
private static final long OPERATION_TIMEOUT = 30000;
25+
private static final int MAX_SUBSCRIPTIONS = 1024;
26+
private final String serverId;
27+
private final IMqttAsyncClient client;
28+
private final MqttConnectOptions connectOptions;
29+
private final ScheduledThreadPoolExecutor recoveryExecutor;
30+
private final Object lifecycle = new Object();
31+
private final LinkedHashMap<String, Subscription> subscriptions = new LinkedHashMap<>();
32+
private ScheduledFuture<?> recovery;
33+
private boolean recovering;
34+
private boolean recoveryRequested;
35+
private boolean wanted = true;
36+
private boolean closed;
37+
private long generation;
38+
private long retrySeconds = 1;
39+
40+
public MqttServerComm(String serverId, String brokerUrl, MqttConnectOptions opts) throws MqttException {
41+
this(serverId, new MqttAsyncClient(brokerUrl, serverId), opts, worker(serverId));
42+
}
43+
44+
public MqttServerComm(String serverId, String brokerUrl, String username, String password) throws MqttException {
45+
this(serverId, brokerUrl, options(username, password));
46+
}
47+
48+
private static MqttConnectOptions options(String username, String password) {
49+
MqttConnectOptions options = new MqttConnectOptions();
50+
options.setCleanSession(true);
51+
if (username != null && !username.isEmpty()) options.setUserName(username);
52+
if (password != null && !password.isEmpty()) options.setPassword(password.toCharArray());
53+
return options;
54+
}
55+
56+
private static ScheduledThreadPoolExecutor worker(String serverId) {
57+
ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(1, runnable -> {
58+
Thread thread = new Thread(runnable, "simpleapi-mqtt-recovery");
59+
thread.setDaemon(true);
60+
return thread;
61+
});
62+
executor.setRemoveOnCancelPolicy(true);
63+
executor.setExecuteExistingDelayedTasksAfterShutdownPolicy(false);
64+
executor.setKeepAliveTime(1, TimeUnit.SECONDS);
65+
executor.allowCoreThreadTimeOut(true);
66+
return executor;
67+
}
68+
69+
// Package-private injection for deterministic lifecycle tests; production uses Paho.
70+
MqttServerComm(String serverId, IMqttAsyncClient client, MqttConnectOptions options,
71+
ScheduledThreadPoolExecutor executor) throws MqttException {
72+
this.serverId = serverId;
73+
this.client = Objects.requireNonNull(client);
74+
this.connectOptions = Objects.requireNonNull(options);
75+
this.recoveryExecutor = Objects.requireNonNull(executor);
76+
// Own reconnect scheduling: Paho's automatic timer has no public cancellation
77+
// contract for disconnect() invoked while its client is already disconnected.
78+
options.setAutomaticReconnect(false);
79+
client.setCallback(new MqttCallbackExtended() {
80+
@Override public void connectComplete(boolean reconnect, String serverURI) { requestRecovery(0); }
81+
@Override public void connectionLost(Throwable cause) {
82+
synchronized (lifecycle) { recoveryRequested = true; }
83+
requestRecovery(1);
84+
}
85+
@Override public void messageArrived(String topic, MqttMessage message) { }
86+
@Override public void deliveryComplete(IMqttDeliveryToken token) { }
87+
});
88+
try { connect(); }
89+
catch (MqttException | RuntimeException failure) {
90+
try { close(); } catch (MqttException cleanup) { failure.addSuppressed(cleanup); }
91+
throw failure;
92+
}
93+
}
94+
95+
public String getServerId() { return serverId; }
96+
public boolean isConnected() { return client.isConnected(); }
97+
98+
public void connect() throws MqttException {
99+
IMqttToken token;
100+
synchronized (lifecycle) {
101+
requireOpen();
102+
wanted = true;
103+
token = client.isConnected() ? null : client.connect(connectOptions);
104+
}
105+
if (token != null) token.waitForCompletion(OPERATION_TIMEOUT);
106+
synchronized (lifecycle) { stopUnwantedConnection(); }
107+
requestRecovery(0);
108+
}
109+
110+
public void disconnect() throws MqttException {
111+
synchronized (lifecycle) {
112+
wanted = false;
113+
generation++;
114+
if (recovery != null) recovery.cancel(false);
115+
recovery = null;
116+
}
117+
if (client.isConnected()) client.disconnect().waitForCompletion(5000);
118+
else client.disconnectForcibly(0, 1000);
119+
}
120+
121+
@Override public void close() throws MqttException {
122+
synchronized (lifecycle) { if (closed) return; closed = true; }
123+
try { disconnect(); }
124+
finally {
125+
recoveryExecutor.shutdownNow();
126+
if (client instanceof MqttAsyncClient paho) paho.close(true);
127+
else client.close();
128+
}
129+
}
130+
131+
public void publish(String topic, String payload, int qos, boolean retained) throws MqttException {
132+
MqttMessage message = new MqttMessage(payload.getBytes(StandardCharsets.UTF_8));
133+
message.setQos(qos);
134+
message.setRetained(retained);
135+
client.publish(topic, message);
136+
}
137+
138+
public void publishEnvelope(String topic, JsonEnvelope envelope, int qos, boolean retained) throws MqttException {
139+
publish(topic, JsonEnvelopeCodec.encode(envelope), qos, retained);
140+
}
141+
142+
public void subscribe(String topicFilter, int qos, MessageListener listener) throws MqttException {
143+
Objects.requireNonNull(topicFilter); Objects.requireNonNull(listener);
144+
if (qos < 0 || qos > 2) throw new IllegalArgumentException("QoS must be 0, 1, or 2");
145+
synchronized (lifecycle) {
146+
requireOpen();
147+
if (!subscriptions.containsKey(topicFilter) && subscriptions.size() >= MAX_SUBSCRIPTIONS)
148+
throw new IllegalStateException("MQTT subscription capacity reached");
149+
Subscription subscription = new Subscription(topicFilter, qos, listener);
150+
subscriptions.put(topicFilter, subscription);
151+
try {
152+
client.subscribe(topicFilter, qos, null, new org.eclipse.paho.client.mqttv3.IMqttActionListener() {
153+
@Override public void onSuccess(IMqttToken token) { }
154+
@Override public void onFailure(IMqttToken token, Throwable failure) {
155+
synchronized (lifecycle) { recoveryRequested = true; }
156+
requestRecovery(1);
157+
}
158+
}, (topic, message) -> listener.messageArrived(topic, message));
159+
}
160+
catch (MqttException failure) { requestRecovery(1); throw failure; }
161+
}
162+
}
163+
164+
public void subscribeEnvelopes(String topicFilter, int qos, EnvelopeListener listener) throws MqttException {
165+
subscribe(topicFilter, qos, (topic, message) -> listener.envelopeArrived(topic,
166+
JsonEnvelopeCodec.decode(new String(message.getPayload(), StandardCharsets.UTF_8))));
167+
}
168+
169+
public void unsubscribe(String topicFilter) throws MqttException {
170+
synchronized (lifecycle) {
171+
requireOpen();
172+
subscriptions.remove(topicFilter);
173+
client.unsubscribe(topicFilter);
174+
}
175+
}
176+
177+
private IMqttToken subscribeNow(Subscription subscription) throws MqttException {
178+
return client.subscribe(subscription.topic(), subscription.qos(), (topic, message) ->
179+
subscription.listener().messageArrived(topic, message));
180+
}
181+
182+
private void requireOpen() throws MqttException {
183+
if (closed) throw new MqttException(MqttException.REASON_CODE_CLIENT_CLOSED);
184+
}
185+
186+
private void requestRecovery(long delay) {
187+
synchronized (lifecycle) {
188+
if (closed || !wanted || recovering || recovery != null) return;
189+
try { recovery = recoveryExecutor.schedule(this::recover, delay, TimeUnit.SECONDS); }
190+
catch (java.util.concurrent.RejectedExecutionException rejected) {
191+
// Only close shuts down this owned executor. Desired subscriptions remain owned.
192+
if (!closed) throw rejected;
193+
}
194+
}
195+
}
196+
197+
private void recover() {
198+
long admitted;
199+
synchronized (lifecycle) {
200+
recovery = null;
201+
if (closed || !wanted) return;
202+
recovering = true;
203+
recoveryRequested = false;
204+
admitted = generation;
205+
}
206+
boolean success = false;
207+
try {
208+
IMqttToken connection;
209+
Subscription[] desired;
210+
synchronized (lifecycle) {
211+
if (!current(admitted)) return;
212+
connection = client.isConnected() ? null : client.connect(connectOptions);
213+
}
214+
if (connection != null) connection.waitForCompletion(OPERATION_TIMEOUT);
215+
synchronized (lifecycle) {
216+
if (!current(admitted)) return;
217+
desired = subscriptions.values().toArray(Subscription[]::new);
218+
}
219+
for (Subscription subscription : desired) {
220+
IMqttToken token;
221+
synchronized (lifecycle) {
222+
if (!current(admitted)) return;
223+
if (subscriptions.get(subscription.topic()) != subscription) continue;
224+
// Admission is serialized with unsubscribe; only token waiting is outside the lock.
225+
token = subscribeNow(subscription);
226+
}
227+
token.waitForCompletion(OPERATION_TIMEOUT);
228+
}
229+
success = client.isConnected();
230+
} catch (MqttException failure) {
231+
// Retry both connection and SUBACK failures with bounded backoff.
232+
} finally {
233+
synchronized (lifecycle) {
234+
recovering = false;
235+
stopUnwantedConnection();
236+
if (success && current(admitted) && !recoveryRequested) retrySeconds = 1;
237+
else {
238+
long delay = retrySeconds;
239+
retrySeconds = Math.min(60, retrySeconds * 2);
240+
requestRecovery(delay);
241+
}
242+
}
243+
}
244+
}
245+
246+
// A CONNECT admitted before disconnect may finish after cancellation. Its
247+
// completion must not silently resurrect a deliberately stopped transport.
248+
private void stopUnwantedConnection() {
249+
if (!wanted && !closed && client.isConnected()) {
250+
try { client.disconnect(); }
251+
catch (MqttException failure) {
252+
// Force the local connection closed if orderly admission is unavailable.
253+
try { client.disconnectForcibly(0, 1000); }
254+
catch (MqttException cleanup) { failure.addSuppressed(cleanup); }
255+
}
256+
}
257+
}
258+
259+
private boolean current(long admitted) { return !closed && wanted && generation == admitted; }
260+
private record Subscription(String topic, int qos, MessageListener listener) { }
261+
public interface MessageListener { void messageArrived(String topic, MqttMessage message) throws Exception; }
262+
public interface EnvelopeListener { void envelopeArrived(String topic, JsonEnvelope envelope) throws Exception; }
114263
}

0 commit comments

Comments
 (0)