From 163858e47b42e47288b9a9de9d2a4a547152e43a Mon Sep 17 00:00:00 2001 From: andsel Date: Sun, 27 Dec 2015 13:19:16 +0100 Subject: [PATCH] Fixed bug in deletion of valid conn descriptor once different clients with same ID connects (issues #116 and #124) --- ChangeLog.txt | 1 + .../moquette/server/ConnectionDescriptor.java | 19 +++++++++ .../moquette/server/netty/NettyChannel.java | 16 ++++++++ .../server/netty/NettyMQTTHandler.java | 2 +- .../moquette/spi/impl/ProtocolProcessor.java | 12 ++++-- .../server/ServerIntegrationPahoTest.java | 39 +++++++++++++++++++ .../java/io/moquette/server/TestCallback.java | 10 +++++ 7 files changed, 94 insertions(+), 5 deletions(-) diff --git a/ChangeLog.txt b/ChangeLog.txt index 579ee1b7..6c731876 100644 --- a/ChangeLog.txt +++ b/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 diff --git a/broker/src/main/java/io/moquette/server/ConnectionDescriptor.java b/broker/src/main/java/io/moquette/server/ConnectionDescriptor.java index 4be8955b..7be84ef1 100644 --- a/broker/src/main/java/io/moquette/server/ConnectionDescriptor.java +++ b/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; + } } diff --git a/broker/src/main/java/io/moquette/server/netty/NettyChannel.java b/broker/src/main/java/io/moquette/server/netty/NettyChannel.java index fefa924a..d0041b4a 100644 --- a/broker/src/main/java/io/moquette/server/netty/NettyChannel.java +++ b/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; + } } diff --git a/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java b/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java index a10c53c1..963d43d4 100644 --- a/broker/src/main/java/io/moquette/server/netty/NettyMQTTHandler.java +++ b/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(); } diff --git a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java b/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java index 0fd508b0..70f946c6 100644 --- a/broker/src/main/java/io/moquette/spi/impl/ProtocolProcessor.java +++ b/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 m_clientIDs; + protected ConcurrentMap 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 m_willStore = new ConcurrentHashMap<>(); + private ConcurrentMap 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(); diff --git a/broker/src/test/java/io/moquette/server/ServerIntegrationPahoTest.java b/broker/src/test/java/io/moquette/server/ServerIntegrationPahoTest.java index 07436b12..c56eb722 100644 --- a/broker/src/test/java/io/moquette/server/ServerIntegrationPahoTest.java +++ b/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; + } + } diff --git a/broker/src/test/java/io/moquette/server/TestCallback.java b/broker/src/test/java/io/moquette/server/TestCallback.java index 3b6c3ce3..199f8a1a 100644 --- a/broker/src/test/java/io/moquette/server/TestCallback.java +++ b/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) { } }