fix: retry IotCoreClient subscriptions when greengrass starts offline (#75)

Allows subscriptions in IotCoreClient to be made eventually if greengrass starts offline.

Author: jcosentino11

src/main/java/com/aws/greengrass/mqtt/bridge/clients/IoTCoreClient.java

@@ -44,26 +44,26 @@ public class IoTCoreClient implements MessageClient {
     private Set<String> subscribedIotCoreTopics = ConcurrentHashMap.newKeySet();
     @Getter(AccessLevel.PROTECTED)
     private Set<String> toSubscribeIotCoreTopics = new HashSet<>();
-
-    private Consumer<Message> messageHandler;
+    private volatile Consumer<Message> messageHandler;
     private Future<?> subscribeFuture;
     private final Object subscribeLock = new Object();
-
     private final MqttClient iotMqttClient;
     private final ExecutorService executorService;
 
     private final Consumer<MqttMessage> iotCoreCallback = (message) -> {
         String topic = message.getTopic();
         LOGGER.atTrace().kv(TOPIC, topic).log("Received IoT Core message");
 
-        if (messageHandler == null) {
+        Consumer<Message> handler = messageHandler;
+        if (handler == null) {
             LOGGER.atWarn().kv(TOPIC, topic).log("IoT Core message received but message handler not set");
         } else {
             Message msg = new Message(topic, message.getPayload());
-            messageHandler.accept(msg);
+            handler.accept(msg);
         }
     };
 
+    @Getter(AccessLevel.PACKAGE) // for unit testing
     private final MqttClientConnectionEvents connectionCallbacks = new MqttClientConnectionEvents() {
         @Override
         public void onConnectionInterrupted(int errorCode) {
@@ -93,7 +93,8 @@ public void onConnectionResumed(boolean sessionPresent) {
     public IoTCoreClient(MqttClient iotMqttClient, ExecutorService executorService) {
         this.iotMqttClient = iotMqttClient;
         this.executorService = executorService;
-        iotMqttClient.addToCallbackEvents(connectionCallbacks);
+        // onConnect handler required to handle case when bridge starts offline
+        iotMqttClient.addToCallbackEvents(connectionCallbacks::onConnectionResumed, connectionCallbacks);
     }
 
     /**
@@ -184,15 +185,16 @@ public boolean supportsTopicFilters() {
 
     @SuppressWarnings({"PMD.AvoidCatchingGenericException", "PMD.PreserveStackTrace", "PMD.ExceptionAsFlowControl"})
     private void subscribeToTopicsWithRetry(Set<String> topics) {
-        // retry only if client is connected; skip if offline.
-        // topics left here should be subscribed when the client is back online (onConnectionResumed event)
-        topics.forEach(s -> {
+        for (String topic : topics) {
             try {
                 RetryUtils.runWithRetry(subscribeRetryConfig, () -> {
                     try {
+                        // retry only if client is connected; skip if offline.
+                        // topics left here should be subscribed when the client
+                        // is back online (onConnectionResumed event)
                         if (iotMqttClient.connected()) {
-                            subscribeToIotCore(s);
-                            subscribedIotCoreTopics.add(s);
+                            subscribeToIotCore(topic);
+                            subscribedIotCoreTopics.add(topic);
                         }
                         // useless return
                         return null;
@@ -207,10 +209,11 @@ private void subscribeToTopicsWithRetry(Set<String> topics) {
                 }, "subscribe-iotcore-topic", LOGGER);
             } catch (InterruptedException e) {
                 Thread.currentThread().interrupt();
+                return;
             } catch (Exception e) {
-                LOGGER.atError().kv(TOPIC, s).setCause(e).log("Failed to subscribe to IoTCore topic");
+                LOGGER.atError().kv(TOPIC, topic).setCause(e).log("Failed to subscribe to IoTCore topic");
             }
-        });
+        }
     }
 
     private void subscribeToIotCore(String topic) throws InterruptedException, ExecutionException, TimeoutException {

src/test/java/com/aws/greengrass/mqtt/bridge/clients/IoTCoreClientTest.java

@@ -29,9 +29,9 @@
 import java.util.stream.Collectors;
 
 import static org.hamcrest.MatcherAssert.assertThat;
-import static org.mockito.ArgumentMatchers.any;
 import static org.hamcrest.Matchers.is;
 import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.lenient;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.reset;
@@ -79,19 +79,32 @@ void GIVEN_iotcore_client_started_WHEN_update_subscriptions_THEN_topics_subscrib
     }
 
     @Test
-    void GIVEN_disconnected_iotcore_client_WHEN_update_subscriptions_THEN_topics_not_subscribed() throws Exception {
+    void GIVEN_offline_iotcore_client_WHEN_update_subscriptions_THEN_subscribe_once_online() throws Exception {
         reset(mockIotMqttClient);
-        when(mockIotMqttClient.connected()).thenReturn(false);
+
         IoTCoreClient iotCoreClient = new IoTCoreClient(mockIotMqttClient, executorService);
+
         Set<String> topics = new HashSet<>();
         topics.add("iotcore/topic");
         topics.add("iotcore/topic2");
+
+        // attempt to update subscriptions, this is expected to fail since bridge is offline
+        when(mockIotMqttClient.connected()).thenReturn(false);
         iotCoreClient.updateSubscriptions(topics, message -> {
         });
 
+        // verify no subscriptions were made
         verify(mockIotMqttClient, never()).subscribe(any(SubscribeRequest.class));
         assertThat(iotCoreClient.getSubscribedIotCoreTopics().size(), is(0));
-        // topics left to be subscribed
+        assertThat(iotCoreClient.getToSubscribeIotCoreTopics(),
+                Matchers.containsInAnyOrder("iotcore/topic", "iotcore/topic2"));
+
+        // simulate mqtt connection resume
+        when(mockIotMqttClient.connected()).thenReturn(true);
+        iotCoreClient.getConnectionCallbacks().onConnectionResumed(false);
+
+        // verify subscriptions were made
+        assertThat(iotCoreClient.getSubscribedIotCoreTopics().size(), is(2));
         assertThat(iotCoreClient.getToSubscribeIotCoreTopics(),
                 Matchers.containsInAnyOrder("iotcore/topic", "iotcore/topic2"));
     }