Added keepalive management to MQTTConnection

broker/src/main/java/io/moquette/broker/ISubscriptionsRepository.java

@@ -0,0 +1,12 @@
+package io.moquette.broker;
+
+import io.moquette.spi.impl.subscriptions.Subscription;
+
+import java.util.List;
+
+public interface ISubscriptionsRepository {
+
+    List<Subscription> listAllSubscriptions();
+
+    void addNewSubscription(Subscription subscription);
+}

broker/src/main/java/io/moquette/broker/MQTTConnection.java

@@ -1,13 +1,14 @@
 package io.moquette.broker;
 
-import io.moquette.server.netty.AutoFlushHandler;
 import io.moquette.server.netty.NettyUtils;
 import io.moquette.spi.impl.DebugUtils;
 import io.moquette.spi.impl.subscriptions.Topic;
 import io.moquette.spi.security.IAuthenticator;
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.Channel;
+import io.netty.channel.ChannelPipeline;
 import io.netty.handler.codec.mqtt.*;
+import io.netty.handler.timeout.IdleStateHandler;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -150,6 +151,7 @@ final class MQTTConnection {
             LOG.trace("Binding MQTTConnection (channel: {}) to session", channel);
             sessionRegistry.bindToSession(this, msg, clientId);
 
+            initializeKeepAliveTimeout(channel, msg, clientId);
             setupInflightResender(channel);
 
             NettyUtils.clientID(channel, clientId);
@@ -165,6 +167,25 @@ final class MQTTConnection {
             .addFirst("inflightResender", new InflightResender(5_000, TimeUnit.MILLISECONDS));
     }
 
+    private void initializeKeepAliveTimeout(Channel channel, MqttConnectMessage msg, String clientId) {
+        int keepAlive = msg.variableHeader().keepAliveTimeSeconds();
+        NettyUtils.keepAlive(channel, keepAlive);
+        NettyUtils.cleanSession(channel, msg.variableHeader().isCleanSession());
+        NettyUtils.clientID(channel, clientId);
+        int idleTime = Math.round(keepAlive * 1.5f);
+        setIdleTime(channel.pipeline(), idleTime);
+
+        LOG.debug("Connection has been configured CId={}, keepAlive={}, removeTemporaryQoS2={}, idleTime={}",
+            clientId, keepAlive, msg.variableHeader().isCleanSession(), idleTime);
+    }
+
+    private void setIdleTime(ChannelPipeline pipeline, int idleTime) {
+        if (pipeline.names().contains("idleStateHandler")) {
+            pipeline.remove("idleStateHandler");
+        }
+        pipeline.addFirst("idleStateHandler", new IdleStateHandler(idleTime, 0, 0));
+    }
+
     private boolean isNotProtocolVersion(MqttConnectMessage msg, MqttVersion version) {
         return msg.variableHeader().version() != version.protocolLevel();
     }
@@ -220,7 +241,8 @@ final class MQTTConnection {
             }
             connected = false;
         }
-        channel.close().addListener(CLOSE_ON_FAILURE);
+//?? this breaks test elapseKeepAliveTime
+//        channel.close().addListener(CLOSE_ON_FAILURE);
     }
 
     void sendConnAck(boolean isSessionAlreadyPresent) {
@@ -229,6 +251,10 @@ final class MQTTConnection {
         channel.writeAndFlush(ackMessage);
     }
 
+    boolean isConnected() {
+        return connected;
+    }
+
     void dropConnection() {
         channel.close();
     }
@@ -307,7 +333,7 @@ final class MQTTConnection {
                 final int messageID = msg.variableHeader().packetId();
                 final Session session = sessionRegistry.retrieve(clientId);
                 session.receivedPublishQos2(messageID, msg);
-                postOffice.receivedPublishQos2(this, msg);
+                postOffice.receivedPublishQos2(this, msg, username);
 //                msg.release();
                 break;
             }
@@ -348,6 +374,9 @@ final class MQTTConnection {
     }
 
     void sendIfWritableElseDrop(MqttMessage msg) {
+        if (LOG.isDebugEnabled()) {
+            LOG.debug("OUT {} on channel {}", msg.fixedHeader().messageType(), channel);
+        }
         if (channel.isWritable()) {
             channel.write(msg);
         }

broker/src/main/java/io/moquette/broker/MemorySubscriptionsRepository.java

@@ -0,0 +1,22 @@
+package io.moquette.broker;
+
+import io.moquette.spi.impl.subscriptions.Subscription;
+
+import java.util.Collections;
+import java.util.LinkedList;
+import java.util.List;
+
+public class MemorySubscriptionsRepository implements ISubscriptionsRepository {
+
+    private final List<Subscription> subscriptions = new LinkedList<>();
+
+    @Override
+    public List<Subscription> listAllSubscriptions() {
+        return Collections.unmodifiableList(subscriptions);
+    }
+
+    @Override
+    public void addNewSubscription(Subscription subscription) {
+        subscriptions.add(subscription);
+    }
+}

broker/src/test/java/io/moquette/broker/MQTTConnectionConnectTest.java

@@ -1,9 +1,6 @@
 package io.moquette.broker;
 
-import io.moquette.persistence.MemoryStorageService;
-import io.moquette.spi.ISessionsStore;
 import io.moquette.spi.impl.MockAuthenticator;
-import io.moquette.spi.impl.SessionsRepository;
 import io.moquette.spi.impl.security.PermitAllAuthorizatorPolicy;
 import io.moquette.spi.impl.subscriptions.CTrieSubscriptionDirectory;
 import io.moquette.spi.impl.subscriptions.ISubscriptionsDirectory;
@@ -45,13 +42,11 @@ public class MQTTConnectionConnectTest {
     public void setUp() {
         connMsg = MqttMessageBuilders.connect().protocolVersion(MqttVersion.MQTT_3_1).cleanSession(true);
 
-        MemoryStorageService memStorage = new MemoryStorageService(null, null);
-        ISessionsStore sessionStore = memStorage.sessionsStore();
         mockAuthenticator = new MockAuthenticator(singleton(FAKE_CLIENT_ID), singletonMap(TEST_USER, TEST_PWD));
 
         ISubscriptionsDirectory subscriptions = new CTrieSubscriptionDirectory();
-        SessionsRepository sessionsRepository = new SessionsRepository(sessionStore, null);
-        subscriptions.init(sessionsRepository);
+        ISubscriptionsRepository subscriptionsRepository = new MemorySubscriptionsRepository();
+        subscriptions.init(subscriptionsRepository);
 
         sessionRegistry = new SessionRegistry(subscriptions, ConnectionTestUtils.NO_OBSERVERS_INTERCEPTOR);
         postOffice = new PostOffice(subscriptions, new PermitAllAuthorizatorPolicy(), new MemoryRetainedRepository(),
@@ -146,7 +141,7 @@ public class MQTTConnectionConnectTest {
 
         // Verify
         verify(postOfficeMock).fireWill(any(Session.Will.class));
-        assertFalse("Connection is accepted and therefore should remain open", channel.isOpen());
+        assertFalse("Connection MUST be disconnected", sut.isConnected());
     }
 
     @Test