Added shutdown sequence of the disruptor used in the output part (from ProtocolProcessor down to the cable)

broker/src/main/java/org/eclipse/moquette/server/Server.java

@@ -103,8 +103,8 @@ public class Server {
     
     public void stopServer() {
     	LOG.info("Server stopping...");
-        messaging.stop();
         m_acceptor.close();
+        messaging.stop();
         LOG.info("Server stopped");
     }
     

broker/src/main/java/org/eclipse/moquette/server/netty/NettyAcceptor.java

@@ -46,6 +46,9 @@ import java.util.Properties;
 import javax.net.ssl.KeyManagerFactory;
 import javax.net.ssl.SSLContext;
 import javax.net.ssl.SSLEngine;
+
+import io.netty.util.concurrent.EventExecutor;
+import io.netty.util.concurrent.Future;
 import org.eclipse.moquette.commons.Constants;
 import org.eclipse.moquette.spi.IMessaging;
 import org.eclipse.moquette.parser.netty.MQTTDecoder;
@@ -270,8 +273,20 @@ public class NettyAcceptor implements ServerAcceptor {
         if (m_bossGroup == null) {
             throw new IllegalStateException("Invoked close on an Acceptor that wasn't initialized");
         }
-        m_workerGroup.shutdownGracefully();
-        m_bossGroup.shutdownGracefully();
+        Future workerWaiter = m_workerGroup.shutdownGracefully();
+        Future bossWaiter = m_bossGroup.shutdownGracefully();
+
+        try {
+            workerWaiter.await(100);
+        } catch (InterruptedException iex) {
+            throw new IllegalStateException(iex);
+        }
+
+        try {
+            bossWaiter.await(100);
+        } catch (InterruptedException iex) {
+            throw new IllegalStateException(iex);
+        }
 
         MessageMetrics metrics = m_metricsCollector.computeMetrics();
         LOG.info("Msg read: {}, msg wrote: {}", metrics.messagesRead(), metrics.messagesWrote());

broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java

@@ -19,7 +19,6 @@ import com.lmax.disruptor.EventHandler;
 import com.lmax.disruptor.RingBuffer;
 import com.lmax.disruptor.dsl.Disruptor;
 import java.nio.ByteBuffer;
-import java.text.ParseException;
 import java.util.*;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -63,7 +62,7 @@ import org.slf4j.LoggerFactory;
  * @author andrea
  */
 class ProtocolProcessor implements EventHandler<ValueEvent> {
-    
+
     static final class WillMessage {
         private final String topic;
         private final ByteBuffer payload;
@@ -111,9 +110,10 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
     
     private ExecutorService m_executor;
     private RingBuffer<ValueEvent> m_ringBuffer;
+    private Disruptor<ValueEvent> m_disruptor;
 
     ProtocolProcessor() {}
-    
+
     /**
      * @param subscriptions the subscription store where are stored all the existing
      *  clients subscriptions.
@@ -142,12 +142,17 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
         //init the output ringbuffer
         m_executor = Executors.newFixedThreadPool(1);
 
-        Disruptor<ValueEvent> disruptor = new Disruptor<>(ValueEvent.EVENT_FACTORY, 1024 * 32, m_executor);
-        disruptor.handleEventsWith(this);
-        disruptor.start();
+        /*Disruptor<ValueEvent>*/ m_disruptor = new Disruptor<>(ValueEvent.EVENT_FACTORY, 1024 * 32, m_executor);
+        m_disruptor.handleEventsWith(this);
+        m_disruptor.start();
 
         // Get the ring buffer from the Disruptor to be used for publishing.
-        m_ringBuffer = disruptor.getRingBuffer();
+        m_ringBuffer = m_disruptor.getRingBuffer();
+    }
+
+    void stop() {
+        m_executor.shutdown();
+        m_disruptor.shutdown();
     }
 
     @MQTTMessage(message = ConnectMessage.class)

broker/src/main/java/org/eclipse/moquette/spi/impl/SimpleMessaging.java

@@ -297,14 +297,15 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
     
     private void processStop() {
         LOG.debug("processStop invoked");
+        m_processor.stop();
+
         if (m_interceptor != null) {
             this.m_interceptor.stop();
+            LOG.debug("interceptor firer stopped");
         }
         m_storageService.close();
         LOG.debug("subscription tree {}", subscriptions.dumpTree());
-//        m_eventProcessor.halt();
-//        m_executor.shutdown();
-        
+
         subscriptions = null;
         m_stopLatch.countDown();