Allow explicit shutdown of the ProtocolAdapterManager for testing

Author: codepitbull

hivemq-edge/src/main/java/com/hivemq/HiveMQEdgeMain.java

@@ -34,6 +34,7 @@
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
 import com.hivemq.http.JaxrsHttpServer;
+import org.jetbrains.annotations.VisibleForTesting;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -174,6 +175,11 @@ public void start(final @Nullable EmbeddedExtension embeddedExtension)
 
     public void stop() {
         stopGateway();
+    }
+
+    @VisibleForTesting
+    public void shutdownProtocolAdapters() {
+        injector.protocolAdapterManager().shutdown();
         try {
             Runtime.getRuntime().removeShutdownHook(shutdownThread);
         } catch (final IllegalStateException ignored) {

hivemq-edge/src/main/java/com/hivemq/embedded/internal/EmbeddedHiveMQImpl.java

@@ -213,6 +213,7 @@ private void performStop(
 
             try {
                 hiveMQServer.stop();
+                hiveMQServer.shutdownProtocolAdapters();
             } catch (final Exception ex) {
                 if (desiredState == State.CLOSED) {
                     log.error("Exception during running shutdown hook.", ex);

hivemq-edge/src/main/java/com/hivemq/protocols/ProtocolAdapterManager.java

@@ -152,6 +152,18 @@ public void start() {
         protocolAdapterConfig.registerConsumer(this::refresh);
     }
 
+    @VisibleForTesting
+    public void shutdown() {
+        protocolAdapters.entrySet().stream().forEach(entry -> {
+            try {
+                entry.getValue().stopAsync(true).get();
+            } catch (final InterruptedException | ExecutionException e) {
+                log.error("Exception happened while shutting down adapter: ", e);
+                throw new RuntimeException(e);
+            }
+        });
+    }
+
     public void refresh(final @NotNull List<ProtocolAdapterEntity> configs) {
         executorService.submit(() -> {
             log.info("Refreshing adapters");