Fixed mem leak due to caching of object without hasCode (NettyChannelContext

ChangeLog.txt

@@ -13,6 +13,7 @@ Version 0.7:
     Added handling of allow_anonymous clients (as in mosquitto issue #34)
     Modified the password_file format to use password stored in sha256 and not plain text
     Added ACL file for topic read|write operations.
+    Removed cause of leakage of Netty Sessions (fix issue #39)
 
 Version 0.6:
     Fix issue #42, bad password file path composition

broker/src/main/java/org/eclipse/moquette/server/netty/NettyMQTTHandler.java

@@ -37,7 +37,7 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter {
     
     private static final Logger LOG = LoggerFactory.getLogger(NettyMQTTHandler.class);
     private IMessaging m_messaging;
-    private final Map<ChannelHandlerContext, NettyChannel> m_channelMapper = new HashMap<ChannelHandlerContext, NettyChannel>();
+//    private final Map<ChannelHandlerContext, NettyChannel> m_channelMapper = new HashMap<>();
     
     @Override
     public void channelRead(ChannelHandlerContext ctx, Object message) {
@@ -54,15 +54,15 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter {
                 case PUBREL:
                 case DISCONNECT:
                 case PUBACK:    
-                    NettyChannel channel;
-                    synchronized(m_channelMapper) {
-                        if (!m_channelMapper.containsKey(ctx)) {
-                            m_channelMapper.put(ctx, new NettyChannel(ctx));
-                        }
-                        channel = m_channelMapper.get(ctx);
-                    }
+//                    NettyChannel channel;
+//                    synchronized(m_channelMapper) {
+//                        if (!m_channelMapper.containsKey(ctx)) {
+//                            m_channelMapper.put(ctx, new NettyChannel(ctx));
+//                        }
+//                        channel = m_channelMapper.get(ctx);
+//                    }
                     
-                    m_messaging.handleProtocolMessage(channel, msg);
+                    m_messaging.handleProtocolMessage(new NettyChannel(ctx), msg);
                     break;
                 case PINGREQ:
                     PingRespMessage pingResp = new PingRespMessage();
@@ -76,13 +76,14 @@ public class NettyMQTTHandler extends ChannelInboundHandlerAdapter {
     
     @Override
     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
-        NettyChannel channel = m_channelMapper.get(ctx);
-        String clientID = (String) channel.getAttribute(NettyChannel.ATTR_KEY_CLIENTID);
-        m_messaging.lostConnection(channel, clientID);
+//        NettyChannel channel = m_channelMapper.get(ctx);
+//        String clientID = (String) channel.getAttribute(NettyChannel.ATTR_KEY_CLIENTID);
+        String clientID = (String) NettyUtils.getAttribute(ctx, NettyChannel.ATTR_KEY_CLIENTID);
+        m_messaging.lostConnection(clientID);
         ctx.close(/*false*/);
-        synchronized(m_channelMapper) {
-            m_channelMapper.remove(ctx);
-        }
+//        synchronized(m_channelMapper) {
+//            m_channelMapper.remove(ctx);
+//        }
     }
     
     public void setMessaging(IMessaging messaging) {

broker/src/main/java/org/eclipse/moquette/server/netty/NettyUtils.java

@@ -0,0 +1,50 @@
+/*
+ * Copyright (c) 2012-2014 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.server.netty;
+
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.timeout.IdleStateHandler;
+import io.netty.util.Attribute;
+import io.netty.util.AttributeKey;
+
+/**
+ * Some Netty's channels utilities.
+ *
+ * @author andrea
+ */
+public class NettyUtils {
+
+    public static Object getAttribute(ChannelHandlerContext channel, AttributeKey<Object> key) {
+        Attribute<Object> attr = channel.attr(key);
+        return attr.get();
+    }
+
+    public static void setAttribute(ChannelHandlerContext channel, AttributeKey<Object> key, Object value) {
+        Attribute<Object> attr = channel.attr(key);
+        attr.set(value);
+    }
+
+    public static void setIdleTime(ChannelHandlerContext channel, int idleTime) {
+        if (channel.pipeline().names().contains("idleStateHandler")) {
+            channel.pipeline().remove("idleStateHandler");
+        }
+        if (channel.pipeline().names().contains("idleEventHandler")) {
+            channel.pipeline().remove("idleEventHandler");
+        }
+        channel.pipeline().addFirst("idleStateHandler", new IdleStateHandler(0, 0, idleTime));
+        channel.pipeline().addAfter("idleStateHandler", "idleEventHandler", new MoquetteIdleTimoutHandler());
+    }
+}

broker/src/main/java/org/eclipse/moquette/spi/IMessaging.java

@@ -30,7 +30,7 @@ public interface IMessaging {
 
     void stop();
 
-    void lostConnection(ServerChannel session, String clientID);
+    void lostConnection(String clientID);
 
     void handleProtocolMessage(ServerChannel session, AbstractMessage msg);
 }

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

@@ -561,12 +561,12 @@ class ProtocolProcessor implements EventHandler<ValueEvent> {
     
     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 (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

@@ -122,8 +122,8 @@ public class SimpleMessaging implements IMessaging, EventHandler<ValueEvent> {
     }
     
     @Override
-    public void lostConnection(ServerChannel session, String clientID) {
-        disruptorPublish(new LostConnectionEvent(session, clientID));
+    public void lostConnection(String clientID) {
+        disruptorPublish(new LostConnectionEvent(clientID));
     }
 
     @Override

broker/src/main/java/org/eclipse/moquette/spi/impl/events/LostConnectionEvent.java

@@ -23,11 +23,11 @@ import org.eclipse.moquette.server.ServerChannel;
  * @author andrea
  */
 public class LostConnectionEvent extends MessagingEvent {
-    public final ServerChannel session;
+//    public final ServerChannel session;
     public final String clientID;
 
-    public LostConnectionEvent(ServerChannel session, String clientID) {
-        this.session = session;
+    public LostConnectionEvent(/*ServerChannel session, */String clientID) {
+//        this.session = session;
         this.clientID = clientID;
     }
 }

broker/src/test/java/org/eclipse/moquette/spi/impl/DummyChannelHandlerContext.java

@@ -0,0 +1,7 @@
+package org.eclipse.moquette.spi.impl;
+
+/**
+ * Created by andrea on 6/7/15.
+ */
+public class DummyChannelHandlerContext {
+}

broker/src/test/java/org/eclipse/moquette/spi/impl/ProtocolProcessorTest.java

@@ -634,45 +634,4 @@ 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, true, new PermitAllAuthorizator());
-
-        //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/integration/issue39_loopingConnecter.groovy

@@ -0,0 +1,34 @@
+@GrabResolver(name='Paho', root='https://repo.eclipse.org/content/repositories/paho-releases/')
+@Grab(group='org.eclipse.paho', module='org.eclipse.paho.client.mqttv3', version='1.0.1', ext='jar')
+
+
+import org.eclipse.paho.client.mqttv3.MqttClient
+import org.eclipse.paho.client.mqttv3.MqttMessage
+import org.eclipse.paho.client.mqttv3.MqttException
+import org.eclipse.paho.client.mqttv3.persist.MqttDefaultFilePersistence
+
+/**
+ * A client that connect and disconnect multiple times, to check that on disconnection there isn't
+ * any resource leakage.
+ * */
+
+if (args.size() != 3) {
+    println "usage: <host> <num_serial_connctions> <local client name>"
+    return
+}
+
+String host = args[0]
+long numSerialConnection = args[1] as long
+String clientName = args[2]
+
+String tmpDir = System.getProperty("java.io.tmpdir")
+MqttDefaultFilePersistence dataStore1 = new MqttDefaultFilePersistence(tmpDir + "/" + clientName)
+MqttClient client1 = new MqttClient("tcp://${host}:1883", "ConCli${clientName}", dataStore1)
+
+println "start loppoing to ${numSerialConnection} conn-discon"
+(1..numSerialConnection).each {
+    client1.connect()
+    client1.disconnect()
+    print "."
+}
+println "Done"