From 285e0af923faac5849758947ebfe65214ede1ea4 Mon Sep 17 00:00:00 2001 From: BenCodez <17074231+BenCodez@users.noreply.github.com> Date: Wed, 30 Sep 2026 21:40:16 -0600 Subject: [PATCH] Recover MQTT connections and subscriptions after broker restarts --- .../servercomm/mqtt/MqttHandler.java | 3 +- .../servercomm/mqtt/MqttServerComm.java | 347 +++++++++++++----- .../mqtt/MqttServerCommLifecycleTest.java | 132 +++++++ .../mqtt/MqttServerCommRecoveryTest.java | 93 +++++ docs/shared-libraries.md | 20 + 5 files changed, 494 insertions(+), 101 deletions(-) create mode 100644 SimpleAPI/src/test/java/com/bencodez/simpleapi/servercomm/mqtt/MqttServerCommLifecycleTest.java create mode 100644 SimpleAPI/src/test/java/com/bencodez/simpleapi/servercomm/mqtt/MqttServerCommRecoveryTest.java diff --git a/SimpleAPI/src/main/java/com/bencodez/simpleapi/servercomm/mqtt/MqttHandler.java b/SimpleAPI/src/main/java/com/bencodez/simpleapi/servercomm/mqtt/MqttHandler.java index 8e6507bb..91364a93 100644 --- a/SimpleAPI/src/main/java/com/bencodez/simpleapi/servercomm/mqtt/MqttHandler.java +++ b/SimpleAPI/src/main/java/com/bencodez/simpleapi/servercomm/mqtt/MqttHandler.java @@ -67,8 +67,7 @@ public void connect() throws Exception { } public void disconnect() throws Exception { - mqtt.disconnect(); - scheduler.shutdownNow(); + try { mqtt.close(); } finally { scheduler.shutdownNow(); } } public boolean isConnected() { diff --git a/SimpleAPI/src/main/java/com/bencodez/simpleapi/servercomm/mqtt/MqttServerComm.java b/SimpleAPI/src/main/java/com/bencodez/simpleapi/servercomm/mqtt/MqttServerComm.java index e6b32d86..a3c5ddf4 100644 --- a/SimpleAPI/src/main/java/com/bencodez/simpleapi/servercomm/mqtt/MqttServerComm.java +++ b/SimpleAPI/src/main/java/com/bencodez/simpleapi/servercomm/mqtt/MqttServerComm.java @@ -1,11 +1,17 @@ package com.bencodez.simpleapi.servercomm.mqtt; import java.nio.charset.StandardCharsets; +import java.util.LinkedHashMap; +import java.util.Objects; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.TimeUnit; -import org.eclipse.paho.client.mqttv3.IMqttActionListener; import org.eclipse.paho.client.mqttv3.IMqttAsyncClient; +import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken; import org.eclipse.paho.client.mqttv3.IMqttToken; import org.eclipse.paho.client.mqttv3.MqttAsyncClient; +import org.eclipse.paho.client.mqttv3.MqttCallbackExtended; import org.eclipse.paho.client.mqttv3.MqttConnectOptions; import org.eclipse.paho.client.mqttv3.MqttException; import org.eclipse.paho.client.mqttv3.MqttMessage; @@ -13,102 +19,245 @@ import com.bencodez.simpleapi.servercomm.codec.JsonEnvelope; import com.bencodez.simpleapi.servercomm.codec.JsonEnvelopeCodec; -/** - * Very basic MQTT client wrapper using Eclipse Paho. - * Provides connect/disconnect, publish, subscribe, and serverId access. - * Supports username/password authentication. - * Envelope helpers included. - */ -public class MqttServerComm { - - private final String serverId; - private final IMqttAsyncClient client; - private final MqttConnectOptions connectOptions; - - public MqttServerComm(String serverId, String brokerUrl, MqttConnectOptions opts) throws MqttException { - this.serverId = serverId; - this.client = new MqttAsyncClient(brokerUrl, serverId); - this.connectOptions = opts; - this.client.connect(connectOptions).waitForCompletion(); - } - - public MqttServerComm(String serverId, String brokerUrl, String username, String password) throws MqttException { - this.serverId = serverId; - this.client = new MqttAsyncClient(brokerUrl, serverId); - this.connectOptions = new MqttConnectOptions(); - this.connectOptions.setCleanSession(true); - - if (username != null && !username.isEmpty()) { - this.connectOptions.setUserName(username); - } - if (password != null && !password.isEmpty()) { - this.connectOptions.setPassword(password.toCharArray()); - } - - this.client.connect(connectOptions).waitForCompletion(); - } - - public String getServerId() { - return serverId; - } - - public void connect() throws MqttException { - if (!client.isConnected()) { - client.connect(connectOptions).waitForCompletion(); - } - } - - public void disconnect() throws MqttException { - if (client.isConnected()) { - client.disconnect().waitForCompletion(); - } - } - - public boolean isConnected() { - return client.isConnected(); - } - - public void publish(String topic, String payload, int qos, boolean retained) throws MqttException { - MqttMessage msg = new MqttMessage(payload.getBytes(StandardCharsets.UTF_8)); - msg.setQos(qos); - msg.setRetained(retained); - client.publish(topic, msg); - } - - public void publishEnvelope(String topic, JsonEnvelope envelope, int qos, boolean retained) throws MqttException { - publish(topic, JsonEnvelopeCodec.encode(envelope), qos, retained); - } - - public void subscribe(String topicFilter, int qos, MessageListener listener) throws MqttException { - client.subscribe(topicFilter, qos, null, new IMqttActionListener() { - @Override - public void onSuccess(IMqttToken asyncActionToken) { - } - - @Override - public void onFailure(IMqttToken asyncActionToken, Throwable exception) { - exception.printStackTrace(); - } - }, (topic, msg) -> listener.messageArrived(topic, msg)); - } - - public void subscribeEnvelopes(String topicFilter, int qos, EnvelopeListener listener) throws MqttException { - subscribe(topicFilter, qos, (topic, msg) -> { - String payload = new String(msg.getPayload(), StandardCharsets.UTF_8); - JsonEnvelope env = JsonEnvelopeCodec.decode(payload); - listener.envelopeArrived(topic, env); - }); - } - - public void unsubscribe(String topicFilter) throws MqttException { - client.unsubscribe(topicFilter); - } - - public interface MessageListener { - void messageArrived(String topic, MqttMessage message) throws Exception; - } - - public interface EnvelopeListener { - void envelopeArrived(String topic, JsonEnvelope envelope) throws Exception; - } +/** MQTT communication with owned, bounded reconnect and subscription recovery. */ +public class MqttServerComm implements AutoCloseable { + private static final long OPERATION_TIMEOUT = 30000; + private static final int MAX_SUBSCRIPTIONS = 1024; + private final String serverId; + private final IMqttAsyncClient client; + private final MqttConnectOptions connectOptions; + private final ScheduledThreadPoolExecutor recoveryExecutor; + private final Object lifecycle = new Object(); + private final LinkedHashMap subscriptions = new LinkedHashMap<>(); + private ScheduledFuture recovery; + private boolean recovering; + private boolean recoveryRequested; + private boolean wanted = true; + private boolean closed; + private long generation; + private long retrySeconds = 1; + + public MqttServerComm(String serverId, String brokerUrl, MqttConnectOptions opts) throws MqttException { + this(serverId, new MqttAsyncClient(brokerUrl, serverId), opts, worker(serverId)); + } + + public MqttServerComm(String serverId, String brokerUrl, String username, String password) throws MqttException { + this(serverId, brokerUrl, options(username, password)); + } + + private static MqttConnectOptions options(String username, String password) { + MqttConnectOptions options = new MqttConnectOptions(); + options.setCleanSession(true); + if (username != null && !username.isEmpty()) options.setUserName(username); + if (password != null && !password.isEmpty()) options.setPassword(password.toCharArray()); + return options; + } + + private static ScheduledThreadPoolExecutor worker(String serverId) { + ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(1, runnable -> { + Thread thread = new Thread(runnable, "simpleapi-mqtt-recovery"); + thread.setDaemon(true); + return thread; + }); + executor.setRemoveOnCancelPolicy(true); + executor.setExecuteExistingDelayedTasksAfterShutdownPolicy(false); + executor.setKeepAliveTime(1, TimeUnit.SECONDS); + executor.allowCoreThreadTimeOut(true); + return executor; + } + + // Package-private injection for deterministic lifecycle tests; production uses Paho. + MqttServerComm(String serverId, IMqttAsyncClient client, MqttConnectOptions options, + ScheduledThreadPoolExecutor executor) throws MqttException { + this.serverId = serverId; + this.client = Objects.requireNonNull(client); + this.connectOptions = Objects.requireNonNull(options); + this.recoveryExecutor = Objects.requireNonNull(executor); + // Own reconnect scheduling: Paho's automatic timer has no public cancellation + // contract for disconnect() invoked while its client is already disconnected. + options.setAutomaticReconnect(false); + client.setCallback(new MqttCallbackExtended() { + @Override public void connectComplete(boolean reconnect, String serverURI) { requestRecovery(0); } + @Override public void connectionLost(Throwable cause) { + synchronized (lifecycle) { recoveryRequested = true; } + requestRecovery(1); + } + @Override public void messageArrived(String topic, MqttMessage message) { } + @Override public void deliveryComplete(IMqttDeliveryToken token) { } + }); + try { connect(); } + catch (MqttException | RuntimeException failure) { + try { close(); } catch (MqttException cleanup) { failure.addSuppressed(cleanup); } + throw failure; + } + } + + public String getServerId() { return serverId; } + public boolean isConnected() { return client.isConnected(); } + + public void connect() throws MqttException { + IMqttToken token; + synchronized (lifecycle) { + requireOpen(); + wanted = true; + token = client.isConnected() ? null : client.connect(connectOptions); + } + if (token != null) token.waitForCompletion(OPERATION_TIMEOUT); + synchronized (lifecycle) { stopUnwantedConnection(); } + requestRecovery(0); + } + + public void disconnect() throws MqttException { + synchronized (lifecycle) { + wanted = false; + generation++; + if (recovery != null) recovery.cancel(false); + recovery = null; + } + if (client.isConnected()) client.disconnect().waitForCompletion(5000); + else client.disconnectForcibly(0, 1000); + } + + @Override public void close() throws MqttException { + synchronized (lifecycle) { if (closed) return; closed = true; } + try { disconnect(); } + finally { + recoveryExecutor.shutdownNow(); + if (client instanceof MqttAsyncClient paho) paho.close(true); + else client.close(); + } + } + + public void publish(String topic, String payload, int qos, boolean retained) throws MqttException { + MqttMessage message = new MqttMessage(payload.getBytes(StandardCharsets.UTF_8)); + message.setQos(qos); + message.setRetained(retained); + client.publish(topic, message); + } + + public void publishEnvelope(String topic, JsonEnvelope envelope, int qos, boolean retained) throws MqttException { + publish(topic, JsonEnvelopeCodec.encode(envelope), qos, retained); + } + + public void subscribe(String topicFilter, int qos, MessageListener listener) throws MqttException { + Objects.requireNonNull(topicFilter); Objects.requireNonNull(listener); + if (qos < 0 || qos > 2) throw new IllegalArgumentException("QoS must be 0, 1, or 2"); + synchronized (lifecycle) { + requireOpen(); + if (!subscriptions.containsKey(topicFilter) && subscriptions.size() >= MAX_SUBSCRIPTIONS) + throw new IllegalStateException("MQTT subscription capacity reached"); + Subscription subscription = new Subscription(topicFilter, qos, listener); + subscriptions.put(topicFilter, subscription); + try { + client.subscribe(topicFilter, qos, null, new org.eclipse.paho.client.mqttv3.IMqttActionListener() { + @Override public void onSuccess(IMqttToken token) { } + @Override public void onFailure(IMqttToken token, Throwable failure) { + synchronized (lifecycle) { recoveryRequested = true; } + requestRecovery(1); + } + }, (topic, message) -> listener.messageArrived(topic, message)); + } + catch (MqttException failure) { requestRecovery(1); throw failure; } + } + } + + public void subscribeEnvelopes(String topicFilter, int qos, EnvelopeListener listener) throws MqttException { + subscribe(topicFilter, qos, (topic, message) -> listener.envelopeArrived(topic, + JsonEnvelopeCodec.decode(new String(message.getPayload(), StandardCharsets.UTF_8)))); + } + + public void unsubscribe(String topicFilter) throws MqttException { + synchronized (lifecycle) { + requireOpen(); + subscriptions.remove(topicFilter); + client.unsubscribe(topicFilter); + } + } + + private IMqttToken subscribeNow(Subscription subscription) throws MqttException { + return client.subscribe(subscription.topic(), subscription.qos(), (topic, message) -> + subscription.listener().messageArrived(topic, message)); + } + + private void requireOpen() throws MqttException { + if (closed) throw new MqttException(MqttException.REASON_CODE_CLIENT_CLOSED); + } + + private void requestRecovery(long delay) { + synchronized (lifecycle) { + if (closed || !wanted || recovering || recovery != null) return; + try { recovery = recoveryExecutor.schedule(this::recover, delay, TimeUnit.SECONDS); } + catch (java.util.concurrent.RejectedExecutionException rejected) { + // Only close shuts down this owned executor. Desired subscriptions remain owned. + if (!closed) throw rejected; + } + } + } + + private void recover() { + long admitted; + synchronized (lifecycle) { + recovery = null; + if (closed || !wanted) return; + recovering = true; + recoveryRequested = false; + admitted = generation; + } + boolean success = false; + try { + IMqttToken connection; + Subscription[] desired; + synchronized (lifecycle) { + if (!current(admitted)) return; + connection = client.isConnected() ? null : client.connect(connectOptions); + } + if (connection != null) connection.waitForCompletion(OPERATION_TIMEOUT); + synchronized (lifecycle) { + if (!current(admitted)) return; + desired = subscriptions.values().toArray(Subscription[]::new); + } + for (Subscription subscription : desired) { + IMqttToken token; + synchronized (lifecycle) { + if (!current(admitted)) return; + if (subscriptions.get(subscription.topic()) != subscription) continue; + // Admission is serialized with unsubscribe; only token waiting is outside the lock. + token = subscribeNow(subscription); + } + token.waitForCompletion(OPERATION_TIMEOUT); + } + success = client.isConnected(); + } catch (MqttException failure) { + // Retry both connection and SUBACK failures with bounded backoff. + } finally { + synchronized (lifecycle) { + recovering = false; + stopUnwantedConnection(); + if (success && current(admitted) && !recoveryRequested) retrySeconds = 1; + else { + long delay = retrySeconds; + retrySeconds = Math.min(60, retrySeconds * 2); + requestRecovery(delay); + } + } + } + } + + // A CONNECT admitted before disconnect may finish after cancellation. Its + // completion must not silently resurrect a deliberately stopped transport. + private void stopUnwantedConnection() { + if (!wanted && !closed && client.isConnected()) { + try { client.disconnect(); } + catch (MqttException failure) { + // Force the local connection closed if orderly admission is unavailable. + try { client.disconnectForcibly(0, 1000); } + catch (MqttException cleanup) { failure.addSuppressed(cleanup); } + } + } + } + + private boolean current(long admitted) { return !closed && wanted && generation == admitted; } + private record Subscription(String topic, int qos, MessageListener listener) { } + public interface MessageListener { void messageArrived(String topic, MqttMessage message) throws Exception; } + public interface EnvelopeListener { void envelopeArrived(String topic, JsonEnvelope envelope) throws Exception; } } diff --git a/SimpleAPI/src/test/java/com/bencodez/simpleapi/servercomm/mqtt/MqttServerCommLifecycleTest.java b/SimpleAPI/src/test/java/com/bencodez/simpleapi/servercomm/mqtt/MqttServerCommLifecycleTest.java new file mode 100644 index 00000000..4eb6ea25 --- /dev/null +++ b/SimpleAPI/src/test/java/com/bencodez/simpleapi/servercomm/mqtt/MqttServerCommLifecycleTest.java @@ -0,0 +1,132 @@ +package com.bencodez.simpleapi.servercomm.mqtt; + +import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.ArgumentMatchers.*; +import static org.mockito.Mockito.*; + +import java.util.ArrayDeque; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import org.eclipse.paho.client.mqttv3.*; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; + +class MqttServerCommLifecycleTest { + private static final class Timer extends ScheduledThreadPoolExecutor { + final ArrayDeque tasks = new ArrayDeque<>(); + final ArrayDeque delays = new ArrayDeque<>(); + Timer() { super(1); } + @Override public ScheduledFuture schedule(Runnable task, long delay, TimeUnit unit) { + tasks.add(task); delays.add(unit.toSeconds(delay)); + return mock(ScheduledFuture.class); + } + void runNext() { delays.remove(); tasks.remove().run(); } + } + private static final class Fixture implements AutoCloseable { + final IMqttAsyncClient client = mock(IMqttAsyncClient.class); + final IMqttToken token = mock(IMqttToken.class); + final AtomicBoolean connected = new AtomicBoolean(); + final Timer timer = new Timer(); + final MqttServerComm comm; + final MqttCallbackExtended callback; + Fixture() throws Exception { + when(client.isConnected()).thenAnswer(call -> connected.get()); + when(client.connect(any())).thenAnswer(call -> { connected.set(true); return token; }); + when(client.disconnect()).thenAnswer(call -> { connected.set(false); return token; }); + when(client.subscribe(anyString(), anyInt(), any(IMqttMessageListener.class))).thenReturn(token); + when(client.subscribe(anyString(), anyInt(), isNull(), any(IMqttActionListener.class), any(IMqttMessageListener.class))).thenReturn(token); + comm = new MqttServerComm("node", client, new MqttConnectOptions(), timer); + ArgumentCaptor captured = ArgumentCaptor.forClass(MqttCallback.class); + verify(client).setCallback(captured.capture()); + callback = (MqttCallbackExtended) captured.getValue(); + timer.runNext(); + } + @Override public void close() throws Exception { comm.close(); timer.shutdownNow(); } + } + @Test void connectionLossRestoresOriginalListenersWithoutManualReconnect() throws Exception { + try (Fixture f = new Fixture()) { + MqttServerComm.MessageListener listener = mock(MqttServerComm.MessageListener.class); + f.comm.subscribe("votes", 1, listener); + f.connected.set(false); + f.callback.connectionLost(new Exception()); + f.callback.connectionLost(new Exception()); + assertEquals(1, f.timer.tasks.size()); + f.timer.runNext(); + verify(f.client, times(2)).connect(any()); + ArgumentCaptor restored = ArgumentCaptor.forClass(IMqttMessageListener.class); + verify(f.client).subscribe(eq("votes"), eq(1), restored.capture()); + MqttMessage vote = new MqttMessage(); + restored.getValue().messageArrived("votes", vote); + verify(listener, times(1)).messageArrived("votes", vote); + assertTrue(f.timer.tasks.isEmpty()); + } + } + @Test void failedReconnectBackoffIsBoundedAndOnlyOneTaskIsOwned() throws Exception { + try (Fixture f = new Fixture()) { + f.connected.set(false); + doThrow(new MqttException(32103)).when(f.client).connect(any()); + f.callback.connectionLost(new Exception()); + for (int i = 0; i < 12; i++) { + assertEquals(1, f.timer.tasks.size()); + assertTrue(f.timer.delays.peek() <= 60); + f.timer.runNext(); + } + } + } + @Test void cancelledRetryCannotReconnectAfterDisconnectOrClose() throws Exception { + try (Fixture f = new Fixture()) { + f.connected.set(false); + f.callback.connectionLost(new Exception()); + f.comm.disconnect(); + f.timer.runNext(); // Model already accepted task racing cancellation. + verify(f.client, times(1)).connect(any()); + f.callback.connectionLost(new Exception()); + assertTrue(f.timer.tasks.isEmpty()); + f.comm.close(); + f.callback.connectComplete(true, "ignored"); + assertTrue(f.timer.tasks.isEmpty()); + assertTrue(f.timer.isShutdown()); + } + } + @Test void disconnectDuringConnectCompletionDoesNotResurrectTransport() throws Exception { + try (Fixture f = new Fixture()) { + f.connected.set(false); + f.callback.connectionLost(new Exception()); + doAnswer(call -> { + f.connected.set(false); + f.comm.disconnect(); + f.connected.set(true); // Late CONNECT completion after stop. + return null; + }).when(f.token).waitForCompletion(30000); + f.timer.runNext(); + assertFalse(f.connected.get()); + verify(f.client).disconnect(); + assertTrue(f.timer.tasks.isEmpty()); + } + } + @Test void unsubscribeRemovesRecoveryIntent() throws Exception { + try (Fixture f = new Fixture()) { + f.comm.subscribe("old", 1, (topic, message) -> {}); + f.comm.unsubscribe("old"); + f.connected.set(false); + f.callback.connectionLost(new Exception()); + f.timer.runNext(); + verify(f.client, never()).subscribe(eq("old"), anyInt(), any(IMqttMessageListener.class)); + } + } + @Test void failedSubackIsRetriedAndCapacityDoesNotGrowWithoutBound() throws Exception { + try (Fixture f = new Fixture()) { + f.comm.subscribe("votes", 1, (topic, message) -> {}); + ArgumentCaptor action = ArgumentCaptor.forClass(IMqttActionListener.class); + verify(f.client).subscribe(eq("votes"), eq(1), isNull(), action.capture(), any(IMqttMessageListener.class)); + action.getValue().onFailure(f.token, new Exception()); + f.timer.runNext(); + verify(f.client).subscribe(eq("votes"), eq(1), any(IMqttMessageListener.class)); + for (int i = 1; i < 1024; i++) f.comm.subscribe("topic/" + i, 1, (topic, message) -> {}); + assertThrows(IllegalStateException.class, () -> f.comm.subscribe("overflow", 1, (topic, message) -> {})); + f.comm.subscribe("votes", 2, (topic, message) -> {}); // Replacement remains possible at capacity. + } + } +} diff --git a/SimpleAPI/src/test/java/com/bencodez/simpleapi/servercomm/mqtt/MqttServerCommRecoveryTest.java b/SimpleAPI/src/test/java/com/bencodez/simpleapi/servercomm/mqtt/MqttServerCommRecoveryTest.java new file mode 100644 index 00000000..b7e6e580 --- /dev/null +++ b/SimpleAPI/src/test/java/com/bencodez/simpleapi/servercomm/mqtt/MqttServerCommRecoveryTest.java @@ -0,0 +1,93 @@ +package com.bencodez.simpleapi.servercomm.mqtt; + +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.IOException; +import java.net.ServerSocket; +import java.nio.file.Files; +import java.nio.file.Path; +import java.time.Duration; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import org.junit.jupiter.api.Assumptions; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +/** Loopback broker restart regression. Run with -Dsimpleapi.mqtt.broker=/path/to/mosquitto. */ +class MqttServerCommRecoveryTest { + @TempDir Path directory; + + @Test + void brokerRestartRestoresBidirectionalSubscriptions() throws Exception { + String executable = System.getProperty("simpleapi.mqtt.broker"); + Assumptions.assumeTrue(executable != null && !executable.isBlank(), "ephemeral mosquitto not configured"); + int port = freePort(); + Path config = directory.resolve("mosquitto.conf"); + Files.writeString(config, "listener " + port + " 127.0.0.1\nallow_anonymous true\n"); + Process broker = startBroker(executable, config); + awaitPort(port, Duration.ofSeconds(5)); + try (MqttServerComm first = new MqttServerComm("recovery-first", "tcp://127.0.0.1:" + port, null, null); + MqttServerComm second = new MqttServerComm("recovery-second", "tcp://127.0.0.1:" + port, null, null)) { + CountDownLatch firstReceived = new CountDownLatch(1); + CountDownLatch secondReceived = new CountDownLatch(1); + CountDownLatch firstAfter = new CountDownLatch(1); + CountDownLatch secondAfter = new CountDownLatch(1); + first.subscribe("recovery/second", 1, (topic, message) -> { + if ("after".equals(new String(message.getPayload(), java.nio.charset.StandardCharsets.UTF_8))) firstAfter.countDown(); + else firstReceived.countDown(); + }); + second.subscribe("recovery/first", 1, (topic, message) -> { + if ("after".equals(new String(message.getPayload(), java.nio.charset.StandardCharsets.UTF_8))) secondAfter.countDown(); + else secondReceived.countDown(); + }); + first.publish("recovery/second", "before", 1, false); + second.publish("recovery/first", "before", 1, false); + assertTrue(firstReceived.await(5, TimeUnit.SECONDS)); + assertTrue(secondReceived.await(5, TimeUnit.SECONDS)); + broker.destroy(); + assertTrue(broker.waitFor(5, TimeUnit.SECONDS)); + awaitDisconnected(first, Duration.ofSeconds(5)); + awaitDisconnected(second, Duration.ofSeconds(5)); + broker = startBroker(executable, config); + awaitPort(port, Duration.ofSeconds(5)); + awaitConnected(first, Duration.ofSeconds(10)); + awaitConnected(second, Duration.ofSeconds(10)); + first.publish("recovery/second", "after", 1, true); + second.publish("recovery/first", "after", 1, true); + assertTrue(firstAfter.await(10, TimeUnit.SECONDS)); + assertTrue(secondAfter.await(10, TimeUnit.SECONDS)); + } finally { + broker.destroyForcibly(); + } + } + + private Process startBroker(String executable, Path config) throws IOException { + Process process = new ProcessBuilder(executable, "-c", config.toString()).redirectErrorStream(true).start(); + return process; + } + + private static void awaitDisconnected(MqttServerComm client, Duration timeout) throws InterruptedException { + long deadline = System.nanoTime() + timeout.toNanos(); + while (client.isConnected() && System.nanoTime() < deadline) Thread.sleep(25); + assertTrue(!client.isConnected(), "MQTT disconnect was not observed"); + } + + private static void awaitConnected(MqttServerComm client, Duration timeout) throws InterruptedException { + long deadline = System.nanoTime() + timeout.toNanos(); + while (!client.isConnected() && System.nanoTime() < deadline) Thread.sleep(25); + assertTrue(client.isConnected(), "MQTT client did not reconnect"); + } + + private static int freePort() throws IOException { + try (ServerSocket socket = new ServerSocket(0)) { return socket.getLocalPort(); } + } + + private static void awaitPort(int port, Duration timeout) throws Exception { + long deadline = System.nanoTime() + timeout.toNanos(); + while (System.nanoTime() < deadline) { + try (java.net.Socket socket = new java.net.Socket("127.0.0.1", port)) { return; } + catch (IOException ignored) { Thread.sleep(25); } + } + throw new IOException("broker did not open loopback port " + port); + } +} diff --git a/docs/shared-libraries.md b/docs/shared-libraries.md index 653a8105..b5e90284 100644 --- a/docs/shared-libraries.md +++ b/docs/shared-libraries.md @@ -172,3 +172,23 @@ version. Existing deployment profiles/configuration remain unchanged; attached artifacts are available to the existing publishing lifecycle. This change does not invoke deployment or claim the new classifier is already on Nexus. There is no separate parent or module publication step to maintain. + +## MQTT broker recovery + +`MqttServerComm` owns a single bounded recovery task with reconnect backoff from +one to sixty seconds. After a broker disconnect it reconnects and restores the +registered topic filters, QoS values and listeners, including clean-session +connections. Failed SUBACKs also retry. Explicit `disconnect()` stops recovery; +`connect()` permits reuse, while `close()` permanently releases the client and +recovery executor. `MqttHandler.disconnect()` is terminal and closes both owned +executors. Desired subscriptions are bounded to 1,024 distinct filters. + +Recovery does not make MQTT QoS delivery exactly once. Consumers must continue to +fence vote effects with their stable occurrence IDs and durable receipts. Publish +calls still report disconnected/error states rather than silently acknowledging +unsent messages; existing durable delivery owners remain responsible for retries. + +The deterministic lifecycle tests run in the normal Maven suite. To additionally +exercise real bidirectional restoration after an ephemeral loopback broker +restart, pass `-Dsimpleapi.mqtt.broker=/path/to/mosquitto`. This test never calls +manual reconnect or re-registers subscriptions after the restart.