Support Session Resumption

src/main/java/org/reaktivity/nukleus/mqtt/internal/MqttConfiguration.java

@@ -33,6 +33,7 @@ public class MqttConfiguration extends Configuration
     public static final BooleanPropertyDef SHARED_SUBSCRIPTION_AVAILABLE;
     public static final BooleanPropertyDef NO_LOCAL;
     public static final IntPropertyDef SESSION_EXPIRY_GRACE_PERIOD;
+    public static final PropertyDef<String> CLIENT_ID;
 
     static
     {
@@ -48,6 +49,7 @@ public class MqttConfiguration extends Configuration
         SHARED_SUBSCRIPTION_AVAILABLE = config.property("shared.subscription.available", false);
         NO_LOCAL = config.property("no.local", true);
         SESSION_EXPIRY_GRACE_PERIOD = config.property("session.expiry.grace.period", 30);
+        CLIENT_ID = config.property("client.id");
         MQTT_CONFIG = config;
     }
 
@@ -111,4 +113,9 @@ public int sessionExpiryGracePeriod()
     {
         return SESSION_EXPIRY_GRACE_PERIOD.get(this);
     }
+
+    public String clientId()
+    {
+        return CLIENT_ID.get(this);
+    }
 }

src/main/java/org/reaktivity/nukleus/mqtt/internal/stream/MqttServerFactory.java

@@ -65,16 +65,21 @@
 import static org.reaktivity.nukleus.mqtt.internal.types.stream.MqttDataExFW.Builder.DEFAULT_EXPIRY_INTERVAL;
 import static org.reaktivity.nukleus.mqtt.internal.types.stream.MqttDataExFW.Builder.DEFAULT_FORMAT;
 
+import java.util.ArrayList;
+import java.util.Comparator;
 import java.util.EnumMap;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.UUID;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Consumer;
 import java.util.function.LongFunction;
 import java.util.function.LongSupplier;
 import java.util.function.LongUnaryOperator;
+import java.util.function.Supplier;
 import java.util.function.ToIntFunction;
 
 import javax.json.Json;
@@ -311,6 +316,7 @@ public final class MqttServerFactory implements StreamFactory
     private final byte sharedSubscriptions;
     private final boolean noLocal;
     private final int sessionExpiryGracePeriod;
+    private final Supplier<String16FW> supplyClientId;
 
     private final MqttValidator validator;
 
@@ -399,6 +405,9 @@ public MqttServerFactory(
         this.sessionExpiryGracePeriod = config.sessionExpiryGracePeriod();
         this.encodeBudgetMax = bufferPool.slotCapacity();
         this.validator = new MqttValidator();
+
+        final Optional<String16FW> clientId = Optional.ofNullable(config.clientId()).map(String16FW::new);
+        this.supplyClientId = clientId.isPresent() ? clientId::get : () -> new String16FW(UUID.randomUUID().toString());
     }
 
     @Override
@@ -1477,7 +1486,7 @@ private int onDecodeConnect(
 
                 if (length == 0)
                 {
-                    this.clientId = new String16FW(UUID.randomUUID().toString());
+                    this.clientId = supplyClientId.get();
                     this.assignedClientId = true;
                 }
                 else if (length > MAXIMUM_CLIENT_ID_LENGTH)
@@ -3500,6 +3509,7 @@ private void onApplicationBegin(
                 doApplicationWindowIfNecessary(traceId, authorization);
             }
 
+            // TODO: parse returned data. could be session data (expiry & delivery), will topic, will message, subscription
             private void onApplicationData(
                 DataFW data)
             {
@@ -4262,6 +4272,128 @@ private void publishWillMessage(
         }
     }
 
+    public class ExpiryComparator implements Comparator<MqttServer> {
+
+        @Override
+        public int compare(MqttServer server1, MqttServer server2) {
+            final long expiry1 = server1.sessionStream.sessionExpiresAt;
+            final long expiry2 = server2.sessionStream.sessionExpiresAt;
+
+            if (expiry1 < expiry2)
+            {
+                return -1;
+            }
+
+            return expiry1 == expiry2 ? 0 : 1;
+        }
+    }
+
+    private class MqttSessionExpiryRefresh
+    {
+        private final MessageConsumer application;
+
+        private final List<MqttServer> sessions = new ArrayList<>();
+
+        private int currentExpiryIndex;
+
+        private long routeId;
+        private long initialId;
+        private long replyId;
+        private long budgetId;
+
+        private long authorization;
+
+        private long sessionExpiresId = NO_CANCEL_ID;
+
+        MqttSessionExpiryRefresh(
+            long routeId,
+            long authorization)
+        {
+            this.routeId = routeId;
+            this.authorization = authorization;
+            this.initialId = supplyInitialId.applyAsLong(routeId);
+            this.replyId = supplyReplyId.applyAsLong(initialId);
+            this.application = router.supplyReceiver(initialId);
+        }
+
+        private void addSession(MqttServer server)
+        {
+            sessions.add(server);
+            sessions.sort(new ExpiryComparator());
+        }
+
+        private void onApplicationInitial(
+            int msgTypeId,
+            DirectBuffer buffer,
+            int index,
+            int length)
+        {
+            switch (msgTypeId)
+            {
+            case SignalFW.TYPE_ID:
+                final SignalFW signal = signalRO.wrap(buffer, index, index + length);
+                onApplicationSignal(signal);
+                break;
+            }
+        }
+
+        private void onApplicationSignal(
+            SignalFW signal)
+        {
+            final int signalId = signal.signalId();
+
+            switch (signalId)
+            {
+            case SESSION_EXPIRY_SIGNAL:
+                onSessionExpiredSignal(signal);
+                break;
+            default:
+                break;
+            }
+        }
+
+        private void onSessionExpiredSignal(
+            SignalFW signal)
+        {
+            final long traceId = signal.traceId();
+            final long authorization = signal.authorization();
+            final long now = System.currentTimeMillis();
+
+            final MqttServer server = sessions.get(currentExpiryIndex);
+            final MqttServer.MqttSessionStream session = server.sessionStream;
+            final long sessionExpiresAt = session.sessionExpiresAt;
+            final long sessionExpiryInterval = server.sessionExpiryInterval;
+
+            currentExpiryIndex = (currentExpiryIndex + 1) % sessions.size();
+
+            if (MqttState.initialClosing(server.state) && now >= sessionExpiresAt)
+            {
+                session.doApplicationFlushOrEnd(traceId, authorization, NO_FLAGS, PUBLISH_ONLY);
+                // publishWillMessage(traceId);
+            }
+            else
+            {
+                sessionExpiresId = NO_CANCEL_ID;
+                doSignalSessionExpirationIfNecessary(session, sessionExpiryInterval);
+            }
+        }
+
+        private void doSignalSessionExpirationIfNecessary(
+            MqttServer.MqttSessionStream session,
+            long sessionExpiryInterval)
+        {
+            session.sessionExpiresAt = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(sessionExpiryInterval);
+
+            final MqttServer nextSession = sessions.get(currentExpiryIndex);
+
+            if (sessionExpiresId == NO_CANCEL_ID)
+            {
+                sessionExpiresId = signaler.signalAt(nextSession.sessionStream.sessionExpiresAt,
+                    routeId, initialId, SESSION_EXPIRY_SIGNAL);
+            }
+        }
+    }
+
     private static boolean invalidWillQos(
         int flags)
     {

src/test/java/org/reaktivity/nukleus/mqtt/internal/ConnectionIT.java

@@ -18,6 +18,7 @@
 import static java.util.concurrent.TimeUnit.SECONDS;
 import static org.junit.rules.RuleChain.outerRule;
 import static org.reaktivity.nukleus.mqtt.internal.MqttConfiguration.PUBLISH_TIMEOUT;
+import static org.reaktivity.nukleus.mqtt.internal.MqttConfigurationTest.CLIENT_ID_NAME;
 import static org.reaktivity.nukleus.mqtt.internal.MqttConfigurationTest.MAXIMUM_QOS_NAME;
 import static org.reaktivity.nukleus.mqtt.internal.MqttConfigurationTest.NO_LOCAL_NAME;
 import static org.reaktivity.nukleus.mqtt.internal.MqttConfigurationTest.RETAIN_AVAILABLE_NAME;
@@ -1144,4 +1145,19 @@ public void shouldSubscribeMessagesWithTwoTopicsBothExactOneSubscribePacket() th
     {
         k3po.finish();
     }
+
+    @Test
+    @Specification({
+        "${routeExt}/session.present/server/controller",
+        "${client}/connect.when.session.present/client",
+        "${server}/connect.when.session.present/server"})
+    @Configure(name = WILDCARD_SUBSCRIPTION_AVAILABLE_NAME, value = "true")
+    @Configure(name = SHARED_SUBSCRIPTION_AVAILABLE_NAME, value = "true")
+    @Configure(name = MAXIMUM_QOS_NAME, value = "2")
+    @Configure(name = SESSION_EXPIRY_INTERVAL_NAME, value = "10")
+    @Configure(name = CLIENT_ID_NAME, value = "123e4567-e89b-42d3-a456-556642440000")
+    public void shouldConnectWhenSessionPresent() throws Exception
+    {
+        k3po.finish();
+    }
 }

src/test/java/org/reaktivity/nukleus/mqtt/internal/MqttConfigurationTest.java

@@ -16,6 +16,7 @@
 package org.reaktivity.nukleus.mqtt.internal;
 
 import static org.junit.Assert.assertEquals;
+import static org.reaktivity.nukleus.mqtt.internal.MqttConfiguration.CLIENT_ID;
 import static org.reaktivity.nukleus.mqtt.internal.MqttConfiguration.CONNECT_TIMEOUT;
 import static org.reaktivity.nukleus.mqtt.internal.MqttConfiguration.MAXIMUM_QOS;
 import static org.reaktivity.nukleus.mqtt.internal.MqttConfiguration.NO_LOCAL;
@@ -43,6 +44,7 @@ public class MqttConfigurationTest
     static final String SHARED_SUBSCRIPTION_AVAILABLE_NAME = "nukleus.mqtt.shared.subscription.available";
     static final String NO_LOCAL_NAME = "nukleus.mqtt.no.local";
     static final String SESSION_EXPIRY_GRACE_PERIOD_NAME = "nukleus.mqtt.session.expiry.grace.period";
+    static final String CLIENT_ID_NAME = "nukleus.mqtt.client.id";
 
     @Test
     public void shouldVerifyConstants() throws Exception
@@ -58,5 +60,6 @@ public void shouldVerifyConstants() throws Exception
         assertEquals(SHARED_SUBSCRIPTION_AVAILABLE.name(), SHARED_SUBSCRIPTION_AVAILABLE_NAME);
         assertEquals(NO_LOCAL.name(), NO_LOCAL_NAME);
         assertEquals(SESSION_EXPIRY_GRACE_PERIOD.name(), SESSION_EXPIRY_GRACE_PERIOD_NAME);
+        assertEquals(CLIENT_ID.name(), CLIENT_ID_NAME);
     }
 }