Fixed serious bug of client descriptor drop on connectionLost event for antoher client with sae ID (bug notified on email by ZhiPeng Yang)

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);

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);
 }

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.*;
-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.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<ValueEvent> {
     
     private static final Logger LOG = LoggerFactory.getLogger(ProtocolProcessor.class);
     
-    private Map<String, ConnectionDescriptor> m_clientIDs = new HashMap<String, ConnectionDescriptor>();
+    private Map<String, ConnectionDescriptor> 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<String, WillMessage> m_willStore = new HashMap<String, WillMessage>();
+    private Map<String, WillMessage> m_willStore = new HashMap<>();
     
     private ExecutorService m_executor;
     private RingBuffer<ValueEvent> m_ringBuffer;
@@ -163,6 +160,7 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
 
         //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<ValueEvent> {
                 cleanSession(msg.getClientID());
             }
 
-            m_clientIDs.get(msg.getClientID()).getSession().close(false);
+            oldSession.close(false);
-            LOG.info("Found an existing connection with same client ID <{}>, forced to close", msg.getClientID());
+            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<ValueEvent> {
     }
         
     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<ValueEvent> {
         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) {
 

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

@@ -107,8 +107,8 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
     }
     
     @Override
-    public void lostConnection(String clientID) {
+    public void lostConnection(ServerChannel session, String clientID) {
-        disruptorPublish(new LostConnectionEvent(clientID));
+        disruptorPublish(new LostConnectionEvent(session, clientID));
     }
 
     @Override
@@ -145,7 +145,7 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
         } 
         if (evt instanceof LostConnectionEvent) {
             LostConnectionEvent lostEvt = (LostConnectionEvent) evt;
-            m_processor.processConnectionLost(lostEvt.getClientID());
+            m_processor.processConnectionLost(lostEvt);
             return;
         }
         

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{
+public class LostConnectionEvent extends MessagingEvent {
-    private String m_clientID;
+    public final ServerChannel session;
+    public final String clientID;
 
-    public LostConnectionEvent(String clienID) {
+    public LostConnectionEvent(ServerChannel session, String clientID) {
-        m_clientID = clienID;
+        this.session = session;
+        this.clientID = clientID;
     }
-
-    public String getClientID() {
-        return m_clientID;
-    }
-    
 }

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());
+    }
 }

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()
+try {
-println "Client1 disconnected"
+    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"