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

@@ -68,13 +68,15 @@
 import java.util.EnumMap;
 import java.util.HashMap;
 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 +313,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 +402,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 +1483,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)

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);
     }
 }