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 4a02f1b4..f4637933 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 @@ -79,7 +79,7 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter { public void channelInactive(ChannelHandlerContext ctx) throws Exception { NettyChannel channel = m_channelMapper.get(ctx); String clientID = (String) channel.getAttribute(Constants.ATTR_CLIENTID); - m_messaging.lostConnection(clientID); + m_messaging.lostConnection(channel, clientID); ctx.close(/*false*/); synchronized(m_channelMapper) { m_channelMapper.remove(ctx); 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 3bf8f637..8d009db5 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/IMessaging.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/IMessaging.java @@ -30,7 +30,7 @@ public interface IMessaging { void stop(); - void lostConnection(String clientID); + void lostConnection(ServerChannel session, String clientID); void handleProtocolMessage(ServerChannel session, AbstractMessage msg); } 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 b7c9bf2e..401a1222 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 @@ -28,10 +28,7 @@ import java.util.concurrent.Executors; import org.eclipse.moquette.spi.IMatchingCondition; import org.eclipse.moquette.spi.IMessagesStore; import org.eclipse.moquette.spi.ISessionsStore; -import org.eclipse.moquette.spi.impl.events.MessagingEvent; -import org.eclipse.moquette.spi.impl.events.OutputMessagingEvent; -import org.eclipse.moquette.spi.impl.events.PubAckEvent; -import org.eclipse.moquette.spi.impl.events.PublishEvent; +import org.eclipse.moquette.spi.impl.events.*; import org.eclipse.moquette.spi.impl.subscriptions.Subscription; import org.eclipse.moquette.spi.impl.subscriptions.SubscriptionsStore; import static org.eclipse.moquette.parser.netty.Utils.VERSION_3_1; @@ -100,13 +97,13 @@ class ProtocolProcessor implements EventHandler { private static final Logger LOG = LoggerFactory.getLogger(ProtocolProcessor.class); - private Map m_clientIDs = new HashMap(); + private Map m_clientIDs = new HashMap<>(); private SubscriptionsStore subscriptions; private IMessagesStore m_messagesStore; private ISessionsStore m_sessionsStore; private IAuthenticator m_authenticator; //maps clientID to Will testament, if specified on CONNECT - private Map m_willStore = new HashMap(); + private Map m_willStore = new HashMap<>(); private ExecutorService m_executor; private RingBuffer m_ringBuffer; @@ -163,6 +160,7 @@ class ProtocolProcessor implements EventHandler { //if an old client with the same ID already exists close its session. if (m_clientIDs.containsKey(msg.getClientID())) { + LOG.info("Found an existing connection with same client ID <{}>, forcing to close", msg.getClientID()); //clean the subscriptions if the old used a cleanSession = true ServerChannel oldSession = m_clientIDs.get(msg.getClientID()).getSession(); boolean cleanSession = (Boolean) oldSession.getAttribute(Constants.CLEAN_SESSION); @@ -171,8 +169,8 @@ class ProtocolProcessor implements EventHandler { cleanSession(msg.getClientID()); } - m_clientIDs.get(msg.getClientID()).getSession().close(false); - LOG.info("Found an existing connection with same client ID <{}>, forced to close", msg.getClientID()); + oldSession.close(false); + LOG.debug("Existing connection with same client ID <{}>, forced to close", msg.getClientID()); } ConnectionDescriptor connDescr = new ConnectionDescriptor(msg.getClientID(), session, msg.isCleanSession()); @@ -283,7 +281,7 @@ class ProtocolProcessor implements EventHandler { } private void processPublish(String clientID, String topic, QOSType qos, ByteBuffer message, boolean retain, Integer messageID) { - LOG.info("Publish recieved from clientID <{}> on topic <{}> with QoS {}", + LOG.info("Publish received from clientID <{}> on topic <{}> with QoS {}", clientID, topic, qos); String publishKey = null; @@ -527,7 +525,15 @@ class ProtocolProcessor implements EventHandler { LOG.info("Disconnected client <{}> with clean session {}", clientID, cleanSession); } - void processConnectionLost(String clientID) { + void processConnectionLost(LostConnectionEvent evt) { + String clientID = evt.clientID; + if (m_clientIDs.containsKey(clientID)) { + if (!m_clientIDs.get(clientID).getSession().equals(evt.session)) { + LOG.info("Received a lost connection with client <{}> for a not matching session", clientID); + return; + } + } + //If already removed a disconnect message was already processed for this clientID if (m_clientIDs.remove(clientID) != null) { 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 d49d6893..654cc5b3 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 @@ -107,8 +107,8 @@ public class SimpleMessaging implements IMessaging, EventHandler { } @Override - public void lostConnection(String clientID) { - disruptorPublish(new LostConnectionEvent(clientID)); + public void lostConnection(ServerChannel session, String clientID) { + disruptorPublish(new LostConnectionEvent(session, clientID)); } @Override @@ -145,7 +145,7 @@ public class SimpleMessaging implements IMessaging, EventHandler { } if (evt instanceof LostConnectionEvent) { LostConnectionEvent lostEvt = (LostConnectionEvent) evt; - m_processor.processConnectionLost(lostEvt.getClientID()); + m_processor.processConnectionLost(lostEvt); return; } 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 index 72ab28fa..e0dffabc 100644 --- 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 @@ -15,20 +15,19 @@ */ package org.eclipse.moquette.spi.impl.events; +import org.eclipse.moquette.server.ServerChannel; + /** * Used to model the connection lost event * * @author andrea */ -public class LostConnectionEvent extends MessagingEvent{ - private String m_clientID; +public class LostConnectionEvent extends MessagingEvent { + public final ServerChannel session; + public final String clientID; - public LostConnectionEvent(String clienID) { - m_clientID = clienID; + public LostConnectionEvent(ServerChannel session, String clientID) { + this.session = session; + this.clientID = clientID; } - - public String getClientID() { - return m_clientID; - } - } diff --git a/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java b/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java index 69f5d676..9713b0b4 100644 --- a/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java +++ b/broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java @@ -26,6 +26,7 @@ import java.util.concurrent.TimeUnit; import org.eclipse.moquette.spi.IMatchingCondition; import org.eclipse.moquette.spi.IMessagesStore; import org.eclipse.moquette.spi.ISessionsStore; +import org.eclipse.moquette.spi.impl.events.LostConnectionEvent; import org.eclipse.moquette.spi.impl.subscriptions.Subscription; import org.eclipse.moquette.spi.impl.subscriptions.SubscriptionsStore; import static org.eclipse.moquette.parser.netty.Utils.VERSION_3_1_1; @@ -272,13 +273,12 @@ public class ProtocolProcessorTest { //Verify AbstractMessage recvMsg = firstReceiverSession.getMessage(); - assertTrue(recvMsg instanceof ConnAckMessage); ConnAckMessage connAckMsg = (ConnAckMessage) recvMsg; assertTrue(connAckMsg.isSessionPresent()); assertEquals(ConnAckMessage.CONNECTION_ACCEPTED, connAckMsg.getReturnCode()); } - + @Test public void testPublish() throws InterruptedException { final Subscription subscription = new Subscription(FAKE_CLIENT_ID, @@ -533,7 +533,7 @@ public class ProtocolProcessorTest { msg.setPayload(buffer); msg.setRetainFlag(true); m_session.setAttribute(Constants.ATTR_CLIENTID, "Publisher"); - m_processor.processPublish(m_session, msg); + m_processor.processPublish(m_session, msg); //Verify no message is received assertNull(m_receivedMessage); @@ -576,4 +576,45 @@ public class ProtocolProcessorTest { }); assertTrue(messages.isEmpty()); } + + + /** + * Simulate a client1 (FAKE_CLIENT_ID) bound with session1, an event of connection lost + * for client2 (FAKE_CLIENT_ID, same of the client1) but from another session, verify + * the session of client1 is not closed. + * */ + @Test + public void testConnectionLostClosesTheCorrectSession() { + MockReceiverChannel channel1 = new MockReceiverChannel(); + + //init the processor + /*SubscriptionsStore subs = new SubscriptionsStore(); + subs.init(new MemoryStorageService());*/ + SubscriptionsStore subs = mock(SubscriptionsStore.class); + m_processor.init(subs, m_storageService, m_sessionStore, null); + + //simulate a connect from client1 FAKE_CLIENT_ID from channel1 + ConnectMessage connectMessage = new ConnectMessage(); + connectMessage.setProcotolVersion((byte) 3); + connectMessage.setClientID(FAKE_CLIENT_ID); + m_processor.processConnect(channel1, connectMessage); + assertConnectReturnCode(ConnAckMessage.CONNECTION_ACCEPTED, channel1); + //ConnAck received + + //send a connection lost event from an already disconnected client, but with same clientID (FAKE_CLIENT_ID) + //Exercise + DummyChannel channel2 = new DummyChannel(); + LostConnectionEvent lostConnectionEvent = new LostConnectionEvent(channel2, FAKE_CLIENT_ID); + m_processor.processConnectionLost(lostConnectionEvent); + + //Verify no subscriptions were deactivated for the client on session1 + verify(subs, never()).deactivate(anyString()); + } + + private void assertConnectReturnCode(byte expectedReturnCode, MockReceiverChannel receiverSession) { + AbstractMessage recvMsg = receiverSession.getMessage(); + assertTrue(recvMsg instanceof ConnAckMessage); + ConnAckMessage connAckMsg = (ConnAckMessage) recvMsg; + assertEquals(expectedReturnCode, connAckMsg.getReturnCode()); + } } diff --git a/tools_scripts/twoClientsWithSameID.groovy b/tools_scripts/twoClientsWithSameID.groovy index b8ded86a..3269df0e 100644 --- a/tools_scripts/twoClientsWithSameID.groovy +++ b/tools_scripts/twoClientsWithSameID.groovy @@ -42,8 +42,12 @@ print "Publisher is going to publish.." publisher.publish('topic', 'Hello world!!'.bytes, 2, false) println "Ok mate, done!" -client1.disconnect() -println "Client1 disconnected" +try { + client1.disconnect() + println "BAD! Client1 has been able to call disconnect, but it has to be already dropped out" +} catch (Exception ex) { + println "Ok, client1 can't disconnect because it was already disconnected" +} client2.disconnect() println "Client2 disconnected"