diff --git a/broker/src/main/java/org/eclipse/moquette/server/Server.java b/broker/src/main/java/org/eclipse/moquette/server/Server.java index 6df25933..da4b7c25 100644 --- a/broker/src/main/java/org/eclipse/moquette/server/Server.java +++ b/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"); } } diff --git a/broker/src/main/java/org/eclipse/moquette/server/ServerAcceptor.java b/broker/src/main/java/org/eclipse/moquette/server/ServerAcceptor.java index 89983407..1549c6d6 100644 --- a/broker/src/main/java/org/eclipse/moquette/server/ServerAcceptor.java +++ b/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(); } diff --git a/broker/src/main/java/org/eclipse/moquette/server/netty/NettyAcceptor.java b/broker/src/main/java/org/eclipse/moquette/server/netty/NettyAcceptor.java index 231eb473..64ec8dd8 100644 --- a/broker/src/main/java/org/eclipse/moquette/server/netty/NettyAcceptor.java +++ b/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 diff --git a/broker/src/main/java/org/eclipse/moquette/server/netty/NettyMQTTHandler.java b/broker/src/main/java/org/eclipse/moquette/server/netty/NettyMQTTHandler.java index 7c43a82f..183ab305 100644 --- a/broker/src/main/java/org/eclipse/moquette/server/netty/NettyMQTTHandler.java +++ b/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; - } } diff --git a/broker/src/main/java/org/eclipse/moquette/spi/IMessaging.java b/broker/src/main/java/org/eclipse/moquette/spi/IMessaging.java index 6079c433..3099f45e 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/IMessaging.java +++ b/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); } diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/AnnotationSupport.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/AnnotationSupport.java deleted file mode 100644 index d101e1a6..00000000 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/AnnotationSupport.java +++ /dev/null @@ -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 messageClassToMethod = new HashMap(); - 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); - } - } -} diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/MQTTMessage.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/MQTTMessage.java deleted file mode 100644 index 92f64a4d..00000000 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/MQTTMessage.java +++ /dev/null @@ -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(); -} diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java index 67077d16..36dc94f4 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/ProtocolProcessor.java +++ b/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 { +public class ProtocolProcessor implements EventHandler { static final class WillMessage { private final String topic; @@ -155,8 +155,7 @@ class ProtocolProcessor implements EventHandler { 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 { } } - @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 { //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 { * 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 { 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 { //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 { 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 { 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 { * 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 topics = msg.topicFilters(); int messageID = msg.getMessageID(); String clientID = (String) session.getAttribute(NettyChannel.ATTR_KEY_CLIENTID); @@ -625,9 +617,8 @@ class ProtocolProcessor implements EventHandler { 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()); diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/SimpleMessaging.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/SimpleMessaging.java index 45295a3d..98f768f8 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/SimpleMessaging.java +++ b/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 { +public class SimpleMessaging { private static final Logger LOG = LoggerFactory.getLogger(SimpleMessaging.class); private SubscriptionsStore subscriptions; - private RingBuffer m_ringBuffer; private IMessagesStore m_storageService; private ISessionsStore m_sessionsStore; - private ExecutorService m_executor; - private Disruptor 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 { 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 m_disruptor = new Disruptor(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 { } m_interceptor = new BrokerInterceptor(observers); - //List 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 { 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 { 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); - } - } } diff --git a/broker/src/test/java/org/eclipse/moquette/server/netty/NettyMQTTHandlerTest.java b/broker/src/test/java/org/eclipse/moquette/server/netty/NettyMQTTHandlerTest.java deleted file mode 100644 index 2fabd231..00000000 --- a/broker/src/test/java/org/eclipse/moquette/server/netty/NettyMQTTHandlerTest.java +++ /dev/null @@ -1,82 +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.server.netty; - -import io.netty.buffer.ByteBuf; -import io.netty.buffer.Unpooled; -import io.netty.channel.ChannelHandlerContext; -import io.netty.channel.ChannelInboundHandlerAdapter; -import io.netty.channel.embedded.EmbeddedChannel; -import io.netty.handler.codec.ByteToMessageDecoder; -import io.netty.handler.codec.CorruptedFrameException; -import org.eclipse.moquette.parser.netty.ConnectDecoder; -import org.eclipse.moquette.spi.IMessaging; -import org.junit.Test; -import static org.junit.Assert.*; - -import java.util.List; - -import static org.mockito.Mockito.mock; - -/** - * - * @author andrea - */ -public class NettyMQTTHandlerTest { - -// class ConnectDecoderAdapter extends ByteToMessageDecoder { -// -// final ConnectDecoder decoder; -// -// ConnectDecoderAdapter(ConnectDecoder decoder) { -// this.decoder = decoder; -// } -// -// @Override -// protected void decode(ChannelHandlerContext ctx, ByteBuf byteBuf, List list) throws Exception { -// decoder.decode(ctx, byteBuf, list); -// } -// } - - class ThrowingExceptionInboundHandler extends ChannelInboundHandlerAdapter { - @Override - public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { - throw new CorruptedFrameException("Invalid protoName: SUPERFAKE"); - } - } - - /** - * Identified by issue #65 - * */ - @Test - public void testCatchingCorruptedFrameException() { - IMessaging mockMsg = mock(IMessaging.class); - NettyMQTTHandler sut = new NettyMQTTHandler(); - sut.setMessaging(mockMsg); - EmbeddedChannel channel = new EmbeddedChannel(new ThrowingExceptionInboundHandler(), sut); - - //prepare a packet with bad protocol name - ByteBuf input = Unpooled.buffer(); - input.writeByte(0x01); - - //Exercise - assertTrue(channel.isOpen()); - channel.writeInbound(input.readBytes(1)); - - //Verify the handler close the channel - assertFalse(channel.isOpen()); - } -} diff --git a/broker/src/test/java/org/eclipse/moquette/spi/impl/AnnotationSupportTest.java b/broker/src/test/java/org/eclipse/moquette/spi/impl/AnnotationSupportTest.java deleted file mode 100644 index df50a9bc..00000000 --- a/broker/src/test/java/org/eclipse/moquette/spi/impl/AnnotationSupportTest.java +++ /dev/null @@ -1,64 +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 org.eclipse.moquette.proto.messages.ConnAckMessage; -import org.eclipse.moquette.proto.messages.ConnectMessage; -import org.eclipse.moquette.server.ServerChannel; -import static org.junit.Assert.*; -import org.junit.Before; -import org.junit.Test; - -/** - * - * @author andrea - */ -public class AnnotationSupportTest { - - class SimplyAnnotatedFixture { - - @MQTTMessage(message = ConnectMessage.class) - void methodToBeDispatched(ServerChannel session, ConnectMessage msg) { - System.out.println("Just a sample method name"); - methodInvoked = true; - } - } - - SimplyAnnotatedFixture annotationTarget = new SimplyAnnotatedFixture(); - boolean methodInvoked; - - @Before - public void setUp() { - methodInvoked = false; - } - - @Test - public void testAnnotationDiscovery() { - AnnotationSupport annHelper = new AnnotationSupport(); - annHelper.processAnnotations(annotationTarget); - annHelper.dispatch(null, new ConnectMessage()); - - //Verify - assertTrue(methodInvoked); - } - - @Test(expected = RuntimeException.class) - public void testAnnotationFailsDiscovery() { - AnnotationSupport annHelper = new AnnotationSupport(); - annHelper.processAnnotations(annotationTarget); - annHelper.dispatch(null, new ConnAckMessage()); - } -}