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 eb6a827d..e41f7f46 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 @@ -21,13 +21,11 @@ import io.netty.channel.ChannelInboundHandlerAdapter; import io.netty.handler.codec.CorruptedFrameException; import org.eclipse.moquette.proto.messages.*; -import org.eclipse.moquette.spi.IMessaging; import org.eclipse.moquette.proto.Utils; import static org.eclipse.moquette.proto.messages.AbstractMessage.*; import org.eclipse.moquette.spi.impl.ProtocolProcessor; -import org.eclipse.moquette.spi.impl.events.LostConnectionEvent; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -39,18 +37,12 @@ import org.slf4j.LoggerFactory; 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) { AbstractMessage msg = (AbstractMessage) message; @@ -105,7 +97,7 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter { if (stolenAttr != null && stolenAttr == Boolean.TRUE) { stolen = stolenAttr; } - m_processor.processConnectionLost(new LostConnectionEvent(clientID, stolen)); + m_processor.processConnectionLost(clientID, stolen); } ctx.close(/*false*/); } 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 8b9c1c77..af9be6de 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 @@ -288,7 +288,7 @@ public class ProtocolProcessor { forward2Subscribers(publishEvt); } else if (qos == AbstractMessage.QOSType.LEAST_ONE) { //QoS1 forward2Subscribers(publishEvt); - sendPubAck(new PubAckEvent(messageID, clientID)); + sendPubAck(clientID, messageID); LOG.debug("replying with PubAck to MSG ID {}", messageID); } else if (qos == AbstractMessage.QOSType.EXACTLY_ONCE) { //QoS2 m_messagesStore.storePublishForFuture(publishEvt); @@ -424,11 +424,10 @@ public class ProtocolProcessor { m_clientIDs.get(clientID).session.write(pubRecMessage); } - private void sendPubAck(PubAckEvent evt) { + private void sendPubAck(String clientId, int messageID) { LOG.trace("sendPubAck invoked"); - String clientId = evt.getClientID(); PubAckMessage pubAckMessage = new PubAckMessage(); - pubAckMessage.setMessageID(evt.getMessageId()); + pubAckMessage.setMessageID(messageID); try { if (m_clientIDs == null) { @@ -515,17 +514,16 @@ public class ProtocolProcessor { LOG.info("DISCONNECT client <{}> finished", clientID, cleanSession); } - public void processConnectionLost(LostConnectionEvent evt) { - String clientID = evt.clientID; + public void processConnectionLost(String clientID, boolean sessionStolen) { //If already removed a disconnect message was already processed for this clientID - if (evt.sessionStolen && m_clientIDs.remove(clientID) != null) { + if (sessionStolen && m_clientIDs.remove(clientID) != null) { //de-activate the subscriptions for this ClientID ClientSession clientSession = m_sessionsStore.sessionForClient(clientID); clientSession.deactivate(); LOG.info("Lost connection with client <{}>", clientID); } //publish the Will message (if any) for the clientID - if (!evt.sessionStolen && m_willStore.containsKey(clientID)) { + if (!sessionStolen && m_willStore.containsKey(clientID)) { WillMessage will = m_willStore.get(clientID); forwardPublishWill(will, clientID); m_willStore.remove(clientID); 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 291b79a1..8cc15ca7 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 @@ -38,9 +38,7 @@ import static org.eclipse.moquette.commons.Constants.*; * * Singleton class that orchestrate the execution of the protocol. * - * Uses the LMAX Disruptor to serialize the incoming, requests, because it work in a evented fashion; - * the requests income from front Netty connectors and are dispatched to the - * ProtocolProcessor. + * It's main responsibility is instantiate the ProtocolProcessor. * * @author andrea */ @@ -70,12 +68,9 @@ public class SimpleMessaging { return INSTANCE; } - public ProtocolProcessor init(IConfig configProps) { + public ProtocolProcessor init(IConfig props) { subscriptions = new SubscriptionsStore(); - return processInit(configProps); - } - 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; diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/events/LostConnectionEvent.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/events/LostConnectionEvent.java deleted file mode 100644 index 2016b374..00000000 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/events/LostConnectionEvent.java +++ /dev/null @@ -1,35 +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.events; - -/** - * Used to model the connection lost event - * - * @author andrea - */ -public class LostConnectionEvent extends MessagingEvent { - public final String clientID; - public final boolean sessionStolen; - - public LostConnectionEvent(String clientID) { - this(clientID, false); - } - - public LostConnectionEvent(String clientID, boolean stolen) { - this.clientID = clientID; - this.sessionStolen = stolen; - } -} diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/events/MessagingEvent.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/events/MessagingEvent.java deleted file mode 100644 index 05a116be..00000000 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/events/MessagingEvent.java +++ /dev/null @@ -1,25 +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.events; - -/** - *Event used to push work request on the SimpleMessaging. - * - * @author andrea - */ -public abstract class MessagingEvent { - -} diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/events/OutputMessagingEvent.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/events/OutputMessagingEvent.java deleted file mode 100644 index 51840935..00000000 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/events/OutputMessagingEvent.java +++ /dev/null @@ -1,43 +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.events; - -import org.eclipse.moquette.proto.messages.AbstractMessage; -import org.eclipse.moquette.server.ServerChannel; - -/** - * - * @author andrea - */ -public class OutputMessagingEvent extends MessagingEvent { - private ServerChannel m_channel; - private AbstractMessage m_message; - - public OutputMessagingEvent(ServerChannel channel, AbstractMessage message) { - m_channel = channel; - m_message = message; - } - - public ServerChannel getChannel() { - return m_channel; - } - - public AbstractMessage getMessage() { - return m_message; - } - - -} diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/events/ProtocolEvent.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/events/ProtocolEvent.java deleted file mode 100644 index cb561c75..00000000 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/events/ProtocolEvent.java +++ /dev/null @@ -1,46 +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.events; - -import org.eclipse.moquette.proto.Utils; -import org.eclipse.moquette.proto.messages.AbstractMessage; -import org.eclipse.moquette.server.ServerChannel; - -/** - * Event used to carry ProtocolMessages from front handler to event processor - */ -public class ProtocolEvent extends MessagingEvent { - ServerChannel m_session; - AbstractMessage message; - - public ProtocolEvent(ServerChannel session, AbstractMessage message) { - this.m_session = session; - this.message = message; - } - - public ServerChannel getSession() { - return m_session; - } - - public AbstractMessage getMessage() { - return message; - } - - @Override - public String toString() { - return "ProtocolEvent wrapping " + Utils.msgType2String(message.getMessageType()); - } -} diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/events/PubAckEvent.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/events/PubAckEvent.java deleted file mode 100644 index 87ddddef..00000000 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/events/PubAckEvent.java +++ /dev/null @@ -1,39 +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.events; - -/** - * Used to send the ack message back to the client after a publish - */ -public class PubAckEvent extends MessagingEvent { - - int m_messageId; - - String m_clientID; - - public PubAckEvent(int messageID, String clientID) { - m_messageId = messageID ; - m_clientID = clientID; - } - - public int getMessageId() { - return m_messageId; - } - - public String getClientID() { - return m_clientID; - } -} diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/events/PublishEvent.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/events/PublishEvent.java index 775b18f1..410b1979 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/events/PublishEvent.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/events/PublishEvent.java @@ -24,7 +24,7 @@ import java.nio.ByteBuffer; * * @author andrea */ -public class PublishEvent extends MessagingEvent { +public class PublishEvent { String m_topic; QOSType m_qos; //byte[] m_message; diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/events/RepublishEvent.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/events/RepublishEvent.java deleted file mode 100644 index 81ed0b3e..00000000 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/events/RepublishEvent.java +++ /dev/null @@ -1,31 +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.events; - -/** - * Used to re-fire stored QoS1 QoS2 events once a client reconnects. - */ -public class RepublishEvent extends MessagingEvent { - private String m_clientID; - - public RepublishEvent(String clientID) { - this.m_clientID = clientID; - } - - public String getClientID() { - return m_clientID; - } -} diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/events/StopEvent.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/events/StopEvent.java deleted file mode 100644 index 194f2aec..00000000 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/events/StopEvent.java +++ /dev/null @@ -1,25 +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.events; - -/** - * Poison pill to shutdown the event loop - * - * @author andrea - */ -public class StopEvent extends MessagingEvent { - -} diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/events/SubscribeEvent.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/events/SubscribeEvent.java deleted file mode 100644 index 733468fc..00000000 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/events/SubscribeEvent.java +++ /dev/null @@ -1,42 +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.events; - -import org.eclipse.moquette.spi.impl.subscriptions.Subscription; - -/** - * - * @author andrea - */ -public class SubscribeEvent extends MessagingEvent { - - Subscription m_subscription; - - int m_messageID; - - public SubscribeEvent(Subscription subscription, int messageID) { - m_subscription = subscription; - m_messageID = messageID; - } - - public Subscription getSubscription() { - return m_subscription; - } - - public int getMessageID() { - return m_messageID; - } -}