Removed frot ringbuffer (old isolation from Netty's thread and single protocol processor thread)

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

@@ -19,6 +19,7 @@ import org.eclipse.moquette.server.config.FilesystemConfig;
 import org.eclipse.moquette.server.config.IConfig;
 import org.eclipse.moquette.server.config.MemoryConfig;
 import org.eclipse.moquette.server.netty.NettyAcceptor;
+import org.eclipse.moquette.spi.impl.ProtocolProcessor;
 import org.eclipse.moquette.spi.impl.SimpleMessaging;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -37,7 +38,6 @@ public class Server {
     private static final Logger LOG = LoggerFactory.getLogger(Server.class);
     
     private ServerAcceptor m_acceptor;
-    SimpleMessaging messaging;
 
     public static void main(String[] args) throws IOException {
         final Server server = new Server();
@@ -93,17 +93,15 @@ public class Server {
             config.setProperty("intercept.handler", handlerProp);
         }
         LOG.info("Persistent store file: " + config.getProperty(PERSISTENT_STORE_PROPERTY_NAME));
-        messaging = SimpleMessaging.getInstance();
-        messaging.init(config);
+        final ProtocolProcessor processor = SimpleMessaging.getInstance().init(config);
 
         m_acceptor = new NettyAcceptor();
-        m_acceptor.initialize(messaging, config);
+        m_acceptor.initialize(processor, config);
     }
     
     public void stopServer() {
     	LOG.info("Server stopping...");
         m_acceptor.close();
-        messaging.stop();
         LOG.info("Server stopped");
     }
 }

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

@@ -19,6 +19,8 @@ import java.io.IOException;
 
 import org.eclipse.moquette.server.config.IConfig;
 import org.eclipse.moquette.spi.IMessaging;
+import org.eclipse.moquette.spi.impl.ProtocolProcessor;
+import org.eclipse.moquette.spi.impl.SimpleMessaging;
 
 /**
  *
@@ -26,7 +28,7 @@ import org.eclipse.moquette.spi.IMessaging;
  */
 public interface ServerAcceptor {
     
-    void initialize(IMessaging messaging, IConfig props) throws IOException;
+    void initialize(ProtocolProcessor processor, IConfig props) throws IOException;
     
     void close();
 }

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

@@ -55,6 +55,8 @@ import org.eclipse.moquette.parser.netty.MQTTDecoder;
 import org.eclipse.moquette.parser.netty.MQTTEncoder;
 import org.eclipse.moquette.server.ServerAcceptor;
 import org.eclipse.moquette.server.netty.metrics.*;
+import org.eclipse.moquette.spi.impl.ProtocolProcessor;
+import org.eclipse.moquette.spi.impl.SimpleMessaging;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -101,12 +103,13 @@ public class NettyAcceptor implements ServerAcceptor {
     MessageMetricsCollector m_metricsCollector = new MessageMetricsCollector();
 
     @Override
-    public void initialize(IMessaging messaging, IConfig props) throws IOException {
+    public void initialize(ProtocolProcessor processor, IConfig props) throws IOException {
         m_bossGroup = new NioEventLoopGroup();
         m_workerGroup = new NioEventLoopGroup();
+        final NettyMQTTHandler handler = new NettyMQTTHandler(processor);
         
-        initializePlainTCPTransport(messaging, props);
-        initializeWebSocketTransport(messaging, props);
+        initializePlainTCPTransport(handler, props);
+        initializeWebSocketTransport(handler, props);
         String sslTcpPortProp = props.getProperty(Constants.SSL_PORT_PROPERTY_NAME);
         String wssPortProp = props.getProperty(Constants.WSS_PORT_PROPERTY_NAME);
         if (sslTcpPortProp != null || wssPortProp != null) {
@@ -115,8 +118,8 @@ public class NettyAcceptor implements ServerAcceptor {
                 LOG.error("Can't initialize SSLHandler layer! Exiting, check your configuration of jks");
                 return;
             }
-            initializeSSLTCPTransport(messaging, props, sslHandlerFactory);
-            initializeWSSTransport(messaging, props, sslHandlerFactory);
+            initializeSSLTCPTransport(handler, props, sslHandlerFactory);
+            initializeWSSTransport(handler, props, sslHandlerFactory);
         }
     }
 
@@ -150,10 +153,8 @@ public class NettyAcceptor implements ServerAcceptor {
         }
     }
     
-    private void initializePlainTCPTransport(IMessaging messaging, IConfig props) throws IOException {
-        final NettyMQTTHandler handler = new NettyMQTTHandler();
+    private void initializePlainTCPTransport(final NettyMQTTHandler handler, IConfig props) throws IOException {
         final MoquetteIdleTimoutHandler timeoutHandler = new MoquetteIdleTimoutHandler();
-        handler.setMessaging(messaging);
         String host = props.getProperty(Constants.HOST_PROPERTY_NAME);
         int port = Integer.parseInt(props.getProperty(Constants.PORT_PROPERTY_NAME));
         initFactory(host, port, new PipelineInitializer() {
@@ -171,7 +172,7 @@ public class NettyAcceptor implements ServerAcceptor {
         });
     }
     
-    private void initializeWebSocketTransport(IMessaging messaging, IConfig props) throws IOException {
+    private void initializeWebSocketTransport(final NettyMQTTHandler handler, IConfig props) throws IOException {
         String webSocketPortProp = props.getProperty(Constants.WEB_SOCKET_PORT_PROPERTY_NAME);
         if (webSocketPortProp == null) {
             //Do nothing no WebSocket configured
@@ -180,9 +181,7 @@ public class NettyAcceptor implements ServerAcceptor {
         }
         int port = Integer.parseInt(webSocketPortProp);
         
-        final NettyMQTTHandler handler = new NettyMQTTHandler();
         final MoquetteIdleTimoutHandler timeoutHandler = new MoquetteIdleTimoutHandler();
-        handler.setMessaging(messaging);
 
         String host = props.getProperty(Constants.HOST_PROPERTY_NAME);
         initFactory(host, port, new PipelineInitializer() {
@@ -205,7 +204,7 @@ public class NettyAcceptor implements ServerAcceptor {
         });
     }
     
-    private void initializeSSLTCPTransport(IMessaging messaging, IConfig props, final SslHandlerFactory sslHandlerFactory) throws IOException {
+    private void initializeSSLTCPTransport(final NettyMQTTHandler handler, IConfig props, final SslHandlerFactory sslHandlerFactory) throws IOException {
         String sslPortProp = props.getProperty(Constants.SSL_PORT_PROPERTY_NAME);
         if (sslPortProp == null) {
             //Do nothing no SSL configured
@@ -216,9 +215,7 @@ public class NettyAcceptor implements ServerAcceptor {
         int sslPort = Integer.parseInt(sslPortProp);
         LOG.info("Starting SSL on port {}", sslPort);
 
-        final NettyMQTTHandler handler = new NettyMQTTHandler();
         final MoquetteIdleTimoutHandler timeoutHandler = new MoquetteIdleTimoutHandler();
-        handler.setMessaging(messaging);
         String host = props.getProperty(Constants.HOST_PROPERTY_NAME);
         initFactory(host, sslPort, new PipelineInitializer() {
             @Override
@@ -236,7 +233,7 @@ public class NettyAcceptor implements ServerAcceptor {
         });
     }
 
-    private void initializeWSSTransport(IMessaging messaging, IConfig props, final SslHandlerFactory sslHandlerFactory) throws IOException {
+    private void initializeWSSTransport(final NettyMQTTHandler handler, IConfig props, final SslHandlerFactory sslHandlerFactory) throws IOException {
         String sslPortProp = props.getProperty(Constants.WSS_PORT_PROPERTY_NAME);
         if (sslPortProp == null) {
             //Do nothing no SSL configured
@@ -244,9 +241,7 @@ public class NettyAcceptor implements ServerAcceptor {
             return;
         }
         int sslPort = Integer.parseInt(sslPortProp);
-        final NettyMQTTHandler handler = new NettyMQTTHandler();
         final MoquetteIdleTimoutHandler timeoutHandler = new MoquetteIdleTimoutHandler();
-        handler.setMessaging(messaging);
         String host = props.getProperty(Constants.HOST_PROPERTY_NAME);
         initFactory(host, sslPort, new PipelineInitializer() {
             @Override

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

@@ -18,16 +18,16 @@ package org.eclipse.moquette.server.netty;
 import io.netty.channel.ChannelHandler.Sharable;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
-import java.util.HashMap;
-import java.util.Map;
 
 import io.netty.handler.codec.CorruptedFrameException;
-import org.apache.commons.codec.binary.StringUtils;
+import org.eclipse.moquette.proto.messages.*;
 import org.eclipse.moquette.spi.IMessaging;
 import org.eclipse.moquette.proto.Utils;
-import org.eclipse.moquette.proto.messages.AbstractMessage;
+
 import static org.eclipse.moquette.proto.messages.AbstractMessage.*;
-import org.eclipse.moquette.proto.messages.PingRespMessage;
+
+import org.eclipse.moquette.spi.impl.ProtocolProcessor;
+import org.eclipse.moquette.spi.impl.events.LostConnectionEvent;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -40,6 +40,16 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter {
     
     private static final Logger LOG = LoggerFactory.getLogger(NettyMQTTHandler.class);
     private IMessaging m_messaging;
+    private final ProtocolProcessor m_processor;
+
+    private NettyMQTTHandler() {
+        m_processor = null;
+    }
+
+    public NettyMQTTHandler(ProtocolProcessor processor) {
+        m_processor = processor;
+    }
+
 
     @Override
     public void channelRead(ChannelHandlerContext ctx, Object message) {
@@ -48,15 +58,31 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter {
         try {
             switch (msg.getMessageType()) {
                 case CONNECT:
+                    m_processor.processConnect(new NettyChannel(ctx), (ConnectMessage) msg);
+                    break;
                 case SUBSCRIBE:
+                    m_processor.processSubscribe(new NettyChannel(ctx), (SubscribeMessage) msg);
+                    break;
                 case UNSUBSCRIBE:
+                    m_processor.processUnsubscribe(new NettyChannel(ctx), (UnsubscribeMessage) msg);
+                    break;
                 case PUBLISH:
+                    m_processor.processPublish(new NettyChannel(ctx), (PublishMessage) msg);
+                    break;
                 case PUBREC:
+                    m_processor.processPubRec(new NettyChannel(ctx), (PubRecMessage) msg);
+                    break;
                 case PUBCOMP:
+                    m_processor.processPubComp(new NettyChannel(ctx), (PubCompMessage) msg);
+                    break;
                 case PUBREL:
+                    m_processor.processPubRel(new NettyChannel(ctx), (PubRelMessage) msg);
+                    break;
                 case DISCONNECT:
-                case PUBACK:    
-                    m_messaging.handleProtocolMessage(new NettyChannel(ctx), msg);
+                    m_processor.processDisconnect(new NettyChannel(ctx), (DisconnectMessage) msg);
+                    break;
+                case PUBACK:
+                    m_processor.processPubAck(new NettyChannel(ctx), (PubAckMessage) msg);
                     break;
                 case PINGREQ:
                     PingRespMessage pingResp = new PingRespMessage();
@@ -74,7 +100,7 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter {
         if (clientID != null && !clientID.isEmpty()) {
             //if the channel was of a correctly connected client, inform messagin
             //else it was of a not completed CONNECT message
-            m_messaging.lostConnection(clientID);
+            m_processor.processConnectionLost(new LostConnectionEvent(clientID));
         }
         ctx.close(/*false*/);
     }
@@ -89,8 +115,4 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter {
         }
         ctx.close();
     }
-
-    public void setMessaging(IMessaging messaging) {
-        m_messaging = messaging;
-    }
 }

broker/src/main/java/org/eclipse/moquette/spi/IMessaging.java

@@ -15,9 +15,6 @@
  */
 package org.eclipse.moquette.spi;
 
-import org.eclipse.moquette.proto.messages.AbstractMessage;
-import org.eclipse.moquette.server.ServerChannel;
-
 /**
  * Callback interface used to be notified of some events from the input event queue.
  * 
@@ -27,10 +24,4 @@ import org.eclipse.moquette.server.ServerChannel;
  * @author andrea
  */
 public interface IMessaging {
-
-    void stop();
-
-    void lostConnection(String clientID);
-
-    void handleProtocolMessage(ServerChannel session, AbstractMessage msg);
 }

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

@@ -1,69 +0,0 @@
-/*
- * Copyright (c) 2012-2015 The original author or authors
- * ------------------------------------------------------
- * All rights reserved. This program and the accompanying materials
- * are made available under the terms of the Eclipse Public License v1.0
- * and Apache License v2.0 which accompanies this distribution.
- *
- * The Eclipse Public License is available at
- * http://www.eclipse.org/legal/epl-v10.html
- *
- * The Apache License v2.0 is available at
- * http://www.opensource.org/licenses/apache2.0.php
- *
- * You may elect to redistribute this code under either of these licenses.
- */
-package org.eclipse.moquette.spi.impl;
-
-import java.lang.reflect.Method;
-import java.util.HashMap;
-import java.util.Map;
-import org.eclipse.moquette.proto.messages.AbstractMessage;
-import org.eclipse.moquette.server.ServerChannel;
-
-/**
- * Annotation @MQTTMessage helper class.
- * 
- * It has the responsibility to identify the @MQTTMessage method of a given instance 
- * or class and define an easy access to invoke them.
- *  
- * @author andrea
- */
-class AnnotationSupport {
-    
-    private final Map<Class, Method> messageClassToMethod = new HashMap<Class, Method>();
-    private Object targetInstance;
-
-    void processAnnotations(Object instance) {
-        targetInstance = instance;
-        Class instanceCls = instance.getClass();
-        Method[] allMethods = instanceCls.getDeclaredMethods();
-        for (Method m : allMethods) {
-            MQTTMessage annotation = m.getAnnotation(MQTTMessage.class);
-            if (annotation == null) {
-                continue;
-            }
-            Class mqttMessageClass = annotation.message();
-            messageClassToMethod.put(mqttMessageClass, m);
-        }
-    }
-    
-    /**
-     * Dispatch a call to the wrapped target invoking the method that match the 
-     * msg class.
-     * 
-     * @throws RuntimeException if the message can't be routed to any annotated method,
-     * or if there are any errors during invocation.
-     */
-    void dispatch(ServerChannel session, AbstractMessage msg) {
-        Method targetMethod = this.messageClassToMethod.get(msg.getClass());
-        if (targetMethod == null) {
-            throw new RuntimeException("Can't dispatch to any @MQTTMessage marked the message: " + msg);
-        }
-        try {
-            targetMethod.invoke(targetInstance, session, msg);
-        } catch (Exception ex) {
-            throw new RuntimeException(ex);
-        } 
-    }
-}

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

@@ -1,33 +0,0 @@
-/*
- * Copyright (c) 2012-2015 The original author or authors
- * ------------------------------------------------------
- * All rights reserved. This program and the accompanying materials
- * are made available under the terms of the Eclipse Public License v1.0
- * and Apache License v2.0 which accompanies this distribution.
- *
- * The Eclipse Public License is available at
- * http://www.eclipse.org/legal/epl-v10.html
- *
- * The Apache License v2.0 is available at
- * http://www.opensource.org/licenses/apache2.0.php
- *
- * You may elect to redistribute this code under either of these licenses.
- */
-package org.eclipse.moquette.spi.impl;
-
-import java.lang.annotation.Documented;
-import java.lang.annotation.Retention;
-import static java.lang.annotation.ElementType.METHOD;
-import static java.lang.annotation.RetentionPolicy.RUNTIME;
-import java.lang.annotation.Target;
-
-/**
- *
- * @author andrea
- */
-@Documented
-@Target(METHOD)
-@Retention(RUNTIME)
-@interface MQTTMessage {
-    Class message();
-}

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

@@ -61,7 +61,7 @@ import org.slf4j.LoggerFactory;
  * 
  * @author andrea
  */
-class ProtocolProcessor implements EventHandler<ValueEvent> {
+public class ProtocolProcessor implements EventHandler<ValueEvent> {
 
     static final class WillMessage {
         private final String topic;
@@ -155,8 +155,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
         m_disruptor.shutdown();
     }
 
-    @MQTTMessage(message = ConnectMessage.class)
-    void processConnect(ServerChannel session, ConnectMessage msg) {
+    public void processConnect(ServerChannel session, ConnectMessage msg) {
         LOG.debug("CONNECT for client <{}>", msg.getClientID());
         if (msg.getProtocolVersion() != VERSION_3_1 && msg.getProtocolVersion() != VERSION_3_1_1) {
             ConnAckMessage badProto = new ConnAckMessage();
@@ -287,8 +286,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
         }
     }
     
-    @MQTTMessage(message = PubAckMessage.class)
-    void processPubAck(ServerChannel session, PubAckMessage msg) {
+    public void processPubAck(ServerChannel session, PubAckMessage msg) {
         String clientID = (String) session.getAttribute(NettyChannel.ATTR_KEY_CLIENTID);
         int messageID = msg.getMessageID();
         //Remove the message from message store
@@ -304,9 +302,8 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
         //remove also the messages stored of type QoS1/2
         m_messagesStore.dropMessagesInSession(clientID);
     }
-    
-    @MQTTMessage(message = PublishMessage.class)
-    void processPublish(ServerChannel session, PublishMessage msg) {
+
+    public void processPublish(ServerChannel session, PublishMessage msg) {
         LOG.trace("PUB --PUBLISH--> SRV executePublish invoked with {}", msg);
         String clientID = (String) session.getAttribute(NettyChannel.ATTR_KEY_CLIENTID);
         final String topic = msg.getTopicName();
@@ -503,8 +500,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
      * Second phase of a publish QoS2 protocol, sent by publisher to the broker. Search the stored message and publish
      * to all interested subscribers.
      * */
-    @MQTTMessage(message = PubRelMessage.class)
-    void processPubRel(ServerChannel session, PubRelMessage msg) {
+    public void processPubRel(ServerChannel session, PubRelMessage msg) {
         String clientID = (String) session.getAttribute(NettyChannel.ATTR_KEY_CLIENTID);
         int messageID = msg.getMessageID();
         LOG.debug("PUB --PUBREL--> SRV processPubRel invoked for clientID {} ad messageID {}", clientID, messageID);
@@ -531,8 +527,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
         disruptorPublish(new OutputMessagingEvent(m_clientIDs.get(clientID).getSession(), pubCompMessage));
     }
     
-    @MQTTMessage(message = PubRecMessage.class)
-    void processPubRec(ServerChannel session, PubRecMessage msg) {
+    public void processPubRec(ServerChannel session, PubRecMessage msg) {
         String clientID = (String) session.getAttribute(NettyChannel.ATTR_KEY_CLIENTID);
         int messageID = msg.getMessageID();
         //once received a PUBREC reply with a PUBREL(messageID)
@@ -545,9 +540,8 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
         //disruptorPublish(new OutputMessagingEvent(m_clientIDs.get(clientID).getSession(), pubRelMessage));
         session.write(pubRelMessage);
     }
-    
-    @MQTTMessage(message = PubCompMessage.class)
-    void processPubComp(ServerChannel session, PubCompMessage msg) {
+
+    public void processPubComp(ServerChannel session, PubCompMessage msg) {
         String clientID = (String) session.getAttribute(NettyChannel.ATTR_KEY_CLIENTID);
         int messageID = msg.getMessageID();
         LOG.debug("\t\tSRV <--PUBCOMP-- SUB processPubComp invoked for clientID {} ad messageID {}", clientID, messageID);
@@ -555,8 +549,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
         m_messagesStore.cleanTemporaryPublish(clientID, messageID);
     }
     
-    @MQTTMessage(message = DisconnectMessage.class)
-    void processDisconnect(ServerChannel session, DisconnectMessage msg) throws InterruptedException {
+    public void processDisconnect(ServerChannel session, DisconnectMessage msg) throws InterruptedException {
         String clientID = (String) session.getAttribute(NettyChannel.ATTR_KEY_CLIENTID);
         boolean cleanSession = (Boolean) session.getAttribute(NettyChannel.ATTR_KEY_CLEANSESSION);
         if (cleanSession) {
@@ -575,8 +568,8 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
         LOG.info("DISCONNECT client <{}> with clean session {}", clientID, cleanSession);
         m_interceptor.notifyClientDisconnected(clientID);
     }
-    
-    void processConnectionLost(LostConnectionEvent evt) {
+
+    public void processConnectionLost(LostConnectionEvent evt) {
         String clientID = evt.clientID;
         //If already removed a disconnect message was already processed for this clientID
         if (m_clientIDs.remove(clientID) != null) {
@@ -597,8 +590,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
      * Remove the clientID from topic subscription, if not previously subscribed,
      * doesn't reply any error
      */
-    @MQTTMessage(message = UnsubscribeMessage.class)
-    void processUnsubscribe(ServerChannel session, UnsubscribeMessage msg) {
+    public void processUnsubscribe(ServerChannel session, UnsubscribeMessage msg) {
         List<String> topics = msg.topicFilters();
         int messageID = msg.getMessageID();
         String clientID = (String) session.getAttribute(NettyChannel.ATTR_KEY_CLIENTID);
@@ -625,9 +617,8 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
         LOG.info("replying with UnsubAck to MSG ID {}", messageID);
         session.write(ackMessage);
     }
-    
-    @MQTTMessage(message = SubscribeMessage.class)
-    void processSubscribe(ServerChannel session, SubscribeMessage msg) {
+
+    public void processSubscribe(ServerChannel session, SubscribeMessage msg) {
         String clientID = (String) session.getAttribute(NettyChannel.ATTR_KEY_CLIENTID);
         boolean cleanSession = (Boolean) session.getAttribute(NettyChannel.ATTR_KEY_CLEANSESSION);
         LOG.debug("SUBSCRIBE client <{}> packetID {}", clientID, msg.getMessageID());

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

@@ -15,22 +15,12 @@
  */
 package org.eclipse.moquette.spi.impl;
 
-import com.lmax.disruptor.EventHandler;
-import com.lmax.disruptor.RingBuffer;
-import com.lmax.disruptor.dsl.Disruptor;
-import org.HdrHistogram.Histogram;
+import org.eclipse.moquette.commons.Constants;
 import org.eclipse.moquette.interception.InterceptHandler;
-import org.eclipse.moquette.proto.messages.AbstractMessage;
 import org.eclipse.moquette.server.config.IConfig;
-import org.eclipse.moquette.spi.impl.security.*;
-import org.eclipse.moquette.server.ServerChannel;
 import org.eclipse.moquette.spi.IMessagesStore;
-import org.eclipse.moquette.spi.IMessaging;
 import org.eclipse.moquette.spi.ISessionsStore;
-import org.eclipse.moquette.spi.impl.events.LostConnectionEvent;
-import org.eclipse.moquette.spi.impl.events.MessagingEvent;
-import org.eclipse.moquette.spi.impl.events.ProtocolEvent;
-import org.eclipse.moquette.spi.impl.events.StopEvent;
+import org.eclipse.moquette.spi.impl.security.*;
 import org.eclipse.moquette.spi.impl.subscriptions.SubscriptionsStore;
 import org.eclipse.moquette.spi.persistence.MapDBPersistentStore;
 import org.slf4j.Logger;
@@ -42,16 +32,7 @@ import java.lang.reflect.Method;
 import java.text.ParseException;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.TimeUnit;
-import org.eclipse.moquette.commons.Constants;
-
-import static org.eclipse.moquette.commons.Constants.PASSWORD_FILE_PROPERTY_NAME;
-import static org.eclipse.moquette.commons.Constants.PERSISTENT_STORE_PROPERTY_NAME;
-import static org.eclipse.moquette.commons.Constants.ALLOW_ANONYMOUS_PROPERTY_NAME;
-import static org.eclipse.moquette.commons.Constants.ACL_FILE_PROPERTY_NAME;
+import static org.eclipse.moquette.commons.Constants.*;
 
 /**
  *
@@ -63,31 +44,22 @@ import static org.eclipse.moquette.commons.Constants.ACL_FILE_PROPERTY_NAME;
  *
  * @author andrea
  */
-public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
+public class SimpleMessaging {
 
     private static final Logger LOG = LoggerFactory.getLogger(SimpleMessaging.class);
 
     private SubscriptionsStore subscriptions;
 
-    private RingBuffer<ValueEvent> m_ringBuffer;
 
     private IMessagesStore m_storageService;
     private ISessionsStore m_sessionsStore;
 
-    private ExecutorService m_executor;
-    private Disruptor<ValueEvent> m_disruptor;
     private BrokerInterceptor m_interceptor;
 
     private static SimpleMessaging INSTANCE;
     
     private final ProtocolProcessor m_processor = new ProtocolProcessor();
-    private final AnnotationSupport annotationSupport = new AnnotationSupport();
-    private boolean benchmarkEnabled = false;
-    
-    CountDownLatch m_stopLatch;
 
-    Histogram histogram = new Histogram(5);
-    
     private SimpleMessaging() {
     }
 
@@ -98,93 +70,12 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
         return INSTANCE;
     }
 
-    public void init(IConfig configProps) {
+    public ProtocolProcessor init(IConfig configProps) {
         subscriptions = new SubscriptionsStore();
-        m_executor = Executors.newFixedThreadPool(1);
-        m_disruptor = new Disruptor<>(ValueEvent.EVENT_FACTORY, 1024 * 32, m_executor); //128 to break the broker
-        /*Disruptor<ValueEvent> m_disruptor = new Disruptor<ValueEvent>(ValueEvent.EVENT_FACTORY, 1024 * 32, m_executor,
-                ProducerType.MULTI, new BusySpinWaitStrategy());*/
-        m_disruptor.handleEventsWith(this);
-        m_disruptor.start();
-
-        // Get the ring buffer from the Disruptor to be used for publishing.
-        m_ringBuffer = m_disruptor.getRingBuffer();
-
-        annotationSupport.processAnnotations(m_processor);
-        processInit(configProps);
+        return processInit(configProps);
     }
 
-    private void disruptorPublish(MessagingEvent msgEvent) {
-        LOG.debug("disruptorPublish publishing event {}, remaining capacity {}", msgEvent, m_ringBuffer.remainingCapacity());
-        long sequence = m_ringBuffer.next();
-        ValueEvent event = m_ringBuffer.get(sequence);
-        event.setEvent(msgEvent);
-        m_ringBuffer.publish(sequence);
-    }
-    
-    @Override
-    public void lostConnection(String clientID) {
-        disruptorPublish(new LostConnectionEvent(clientID));
-    }
-
-    @Override
-    public void handleProtocolMessage(ServerChannel session, AbstractMessage msg) {
-        disruptorPublish(new ProtocolEvent(session, msg));
-    }
-
-    @Override
-    public void stop() {
-        m_stopLatch = new CountDownLatch(1);
-        disruptorPublish(new StopEvent());
-        try {
-            //wait the callback notification from the protocol processor thread
-            LOG.debug("waiting 10 sec to m_stopLatch");
-            boolean elapsed = !m_stopLatch.await(10, TimeUnit.SECONDS);
-            LOG.debug("after m_stopLatch");
-            m_executor.shutdown();
-            m_disruptor.shutdown();
-            if (elapsed) {
-                LOG.error("Can't stop the server in 10 seconds");
-            }
-        } catch (InterruptedException ex) {
-            LOG.error(null, ex);
-        }
-    }
-    
-    @Override
-    public void onEvent(ValueEvent t, long l, boolean bln) throws Exception {
-        MessagingEvent evt = t.getEvent();
-        t.setEvent(null); //free the reference to all Netty stuff
-        LOG.info("onEvent processing messaging event from input ringbuffer {}", evt);
-        if (evt instanceof StopEvent) {
-            processStop();
-            return;
-        } 
-        if (evt instanceof LostConnectionEvent) {
-            LostConnectionEvent lostEvt = (LostConnectionEvent) evt;
-            m_processor.processConnectionLost(lostEvt);
-            return;
-        }
-        
-        if (evt instanceof ProtocolEvent) {
-            ServerChannel session = ((ProtocolEvent) evt).getSession();
-            AbstractMessage message = ((ProtocolEvent) evt).getMessage();
-            try {
-                long startTime = System.nanoTime();
-                annotationSupport.dispatch(session, message);
-                if (benchmarkEnabled) {
-                    long delay = System.nanoTime() - startTime;
-                    histogram.recordValue(delay);
-                }
-            } catch (Throwable th) {
-                LOG.error("Serious error processing the message {} for {}", message, session, th);
-            }
-        }
-    }
-
-    private void processInit(IConfig props) {
-        benchmarkEnabled = Boolean.parseBoolean(System.getProperty("moquette.processor.benchmark", "false"));
-
+    private ProtocolProcessor processInit(IConfig props) {
         //TODO use a property to select the storage path
         MapDBPersistentStore mapStorage = new MapDBPersistentStore(props.getProperty(PERSISTENT_STORE_PROPERTY_NAME, ""));
         m_storageService = mapStorage;
@@ -204,8 +95,6 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
         }
         m_interceptor = new BrokerInterceptor(observers);
 
-        //List<Subscription> storedSubscriptions = m_sessionsStore.listAllSubscriptions();
-        //subscriptions.init(storedSubscriptions);
         subscriptions.init(m_sessionsStore);
 
         String configPath = System.getProperty("moquette.path", null);
@@ -253,6 +142,7 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
 
         boolean allowAnonymous = Boolean.parseBoolean(props.getProperty(ALLOW_ANONYMOUS_PROPERTY_NAME, "true"));
         m_processor.init(subscriptions, m_storageService, m_sessionsStore, authenticator, allowAnonymous, authorizator, m_interceptor);
+        return m_processor;
     }
     
     private Object loadClass(String className, Class<?> cls) {
@@ -289,24 +179,4 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
 
         return instance;
     }
-    
-    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());
-
-        subscriptions = null;
-        m_stopLatch.countDown();
-
-        if (benchmarkEnabled) {
-            //log metrics
-            histogram.outputPercentileDistribution(System.out, 1000.0);
-        }
-    }
 }