Fixed bug in deletion of valid conn descriptor once different clients with same ID connects (issues #116 and #124)

ChangeLog.txt

@@ -17,6 +17,7 @@ Version 0.8:
     Removed leaking ref to parent's node (issue #119 and #126)
     Added internal publish once the broker is embedded (issue #113)
     Removed client active/not active state from persistence (issue #118)
+    Avoid deletion of client descriptor in case of multiple clients with same clientID (#)
 
 Version 0.7:
     Fix issue #38, added aligned to MQTT 3.1.1

broker/src/main/java/io/moquette/server/ConnectionDescriptor.java

@@ -38,4 +38,23 @@ public class ConnectionDescriptor {
     public String toString() {
         return "ConnectionDescriptor{" + "clientID=" + clientID + ", cleanSession=" + cleanSession + '}';
     }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+
+        ConnectionDescriptor that = (ConnectionDescriptor) o;
+
+        if (clientID != null ? !clientID.equals(that.clientID) : that.clientID != null) return false;
+        return !(session != null ? !session.equals(that.session) : that.session != null);
+
+    }
+
+    @Override
+    public int hashCode() {
+        int result = clientID != null ? clientID.hashCode() : 0;
+        result = 31 * result + (session != null ? session.hashCode() : 0);
+        return result;
+    }
 }

broker/src/main/java/io/moquette/server/netty/NettyChannel.java

@@ -73,4 +73,20 @@ public class NettyChannel implements ServerChannel {
         String clientID = (String) getAttribute(ATTR_KEY_CLIENTID);
         return "session [clientID: "+ clientID +"]";
     }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+
+        NettyChannel that = (NettyChannel) o;
+
+        return !(m_channel != null ? !m_channel.channel().equals(that.m_channel.channel()) : that.m_channel != null);
+
+    }
+
+    @Override
+    public int hashCode() {
+        return m_channel != null ? m_channel.channel().hashCode() : 0;
+    }
 }

broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java

@@ -96,7 +96,7 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter {
             if (stolenAttr != null && stolenAttr == Boolean.TRUE) {
                 stolen = stolenAttr;
             }
-            m_processor.processConnectionLost(clientID, stolen);
+            m_processor.processConnectionLost(clientID, stolen, new NettyChannel(ctx));
         }
         ctx.close();
     }

broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java

@@ -18,6 +18,7 @@ package io.moquette.spi.impl;
 import java.nio.ByteBuffer;
 import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
 
 import io.moquette.server.ConnectionDescriptor;
 import io.moquette.server.ServerChannel;
@@ -92,7 +93,7 @@ public class ProtocolProcessor {
     
     private static final Logger LOG = LoggerFactory.getLogger(ProtocolProcessor.class);
     
-    protected Map<String, ConnectionDescriptor> m_clientIDs;
+    protected ConcurrentMap<String, ConnectionDescriptor> m_clientIDs;
     private SubscriptionsStore subscriptions;
     private boolean allowAnonymous;
     private IAuthorizator m_authorizator;
@@ -102,7 +103,7 @@ public class ProtocolProcessor {
     private BrokerInterceptor m_interceptor;
 
     //maps clientID to Will testament, if specified on CONNECT
-    private Map<String, WillMessage> m_willStore = new ConcurrentHashMap<>();
+    private ConcurrentMap<String, WillMessage> m_willStore = new ConcurrentHashMap<>();
     
     ProtocolProcessor() {}
 
@@ -230,6 +231,7 @@ public class ProtocolProcessor {
             republishStoredInSession(clientSession);
         }
         LOG.info("CONNECT processed");
+//        LOG.info("CONNECT clients descriptors {}", m_clientIDs);
     }
 
     private void failedCredentials(ServerChannel session) {
@@ -630,9 +632,11 @@ public class ProtocolProcessor {
         LOG.info("DISCONNECT client <{}> finished", clientID, cleanSession);
     }
 
-    public void processConnectionLost(String clientID, boolean sessionStolen) {
+    public void processConnectionLost(String clientID, boolean sessionStolen, NettyChannel channel) {
+        ConnectionDescriptor oldConnDescr = new ConnectionDescriptor(clientID, channel, true);
+        m_clientIDs.remove(clientID, oldConnDescr);
         //If already removed a disconnect message was already processed for this clientID
-        if (sessionStolen && m_clientIDs.remove(clientID) != null) {
+        if (sessionStolen) {
             //de-activate the subscriptions for this ClientID
             ClientSession clientSession = m_sessionsStore.sessionForClient(clientID);
             clientSession.deactivate();

broker/src/test/java/io/moquette/server/ServerIntegrationPahoTest.java

@@ -443,4 +443,43 @@ public class ServerIntegrationPahoTest {
         assertEquals("Hello", new String(messageOnB.getPayload()));
     }
 
+    @Test
+    public void testForceClientDisconnection_issue116() throws Exception {
+        LOG.info("*** testForceClientDisconnection_issue118 ***");
+        TestCallback cbSubscriber1 = new TestCallback();
+        MqttClient clientXA = createClient("subscriber", "X", cbSubscriber1);
+        clientXA.subscribe("topic", 0);
+
+        MqttClient clientXB = createClient("publisher", "X");
+        clientXB.publish("topic", "Hello".getBytes(), 2, true);
+
+        TestCallback cbSubscriber2 = new TestCallback();
+        MqttClient clientYA = createClient("subscriber", "Y", cbSubscriber2);
+        clientYA.subscribe("topic", 0);
+
+        MqttClient clientYB = createClient("publisher", "Y");
+        clientYB.publish("topic", "Hello 2".getBytes(), 2, true);
+
+        //Verify that the second subscriber client get notified and not the first.
+        assertTrue(cbSubscriber1.connectionLost());
+        assertEquals("Hello 2", new String(cbSubscriber2.getMessage(true).getPayload()));
+    }
+
+
+    protected MqttClient createClient(String clientName, String storeSuffix) throws MqttException {
+        return createClient(clientName, storeSuffix, null);
+    }
+
+    protected MqttClient createClient(String clientName, String storeSuffix, TestCallback cb) throws MqttException {
+        String tmpDir = System.getProperty("java.io.tmpdir");
+        //clientX connect and subscribe to /topic QoS2
+        MqttClientPersistence dsClient = new MqttDefaultFilePersistence(tmpDir + File.separator + storeSuffix + clientName);
+        MqttClient client = new MqttClient("tcp://localhost:1883", clientName, dsClient);
+        if (cb != null) {
+            client.setCallback(cb);
+        }
+        client.connect();
+        return client;
+    }
+
 }

broker/src/test/java/io/moquette/server/TestCallback.java

@@ -30,6 +30,7 @@ class TestCallback implements MqttCallback {
         private MqttMessage m_message;
         private String m_topic;
         private CountDownLatch m_latch = new CountDownLatch(1);
+        private boolean m_connectionLost = false;
 
         public MqttMessage getMessage(boolean checkElapsed) {
             try {
@@ -56,17 +57,26 @@ class TestCallback implements MqttCallback {
             m_latch = new CountDownLatch(1);
             m_message = null;
             m_topic = null;
+            m_connectionLost = false;
         }
 
+        public boolean connectionLost() {
+            return m_connectionLost;
+        }
+
+        @Override
         public void connectionLost(Throwable cause) {
+            m_connectionLost = true;
         }
 
+        @Override
         public void messageArrived(String topic, MqttMessage message) throws Exception {
             m_message = message;
             m_topic = topic;
             m_latch.countDown();
         }
 
+        @Override
         public void deliveryComplete(IMqttDeliveryToken token) {
         }
     }