Merge pull request #6 from JavaSaBr/feature-broker-5

Implement ClientIdRegistry

Author: crazyrokr

build.gradle

@@ -46,10 +46,12 @@ allprojects {
     }
 
     dependencies {
+    
         implementation "com.spaceshift:rlib.network:$rlibVersion"
         implementation "com.spaceshift:rlib.logger.slf4j:$rlibVersion"
         implementation "org.springframework.boot:spring-boot-starter:$springbootVersion"
         implementation "org.springframework.boot:spring-boot-starter-logging:$springbootVersion"
+        implementation "io.projectreactor:reactor-core:$projectReactorVersion"
 
         compileOnly "org.jetbrains:annotations:$annotationVersion"
         compileOnly "org.projectlombok:lombok:$lombokVersion"

src/main/java/com/ss/mqtt/broker/config/MqttBrokerConfig.java

@@ -3,15 +3,15 @@
 import com.ss.mqtt.broker.model.MqttPropertyConstants;
 import com.ss.mqtt.broker.model.QoS;
 import com.ss.mqtt.broker.network.MqttConnection;
-import com.ss.mqtt.broker.network.packet.in.MqttReadablePacket;
-import com.ss.mqtt.broker.network.packet.out.MqttWritablePacket;
+import com.ss.mqtt.broker.network.client.UnsafeMqttClient;
+import com.ss.mqtt.broker.network.client.impl.DeviceMqttClient;
+import com.ss.mqtt.broker.network.packet.PacketType;
+import com.ss.mqtt.broker.network.packet.in.handler.*;
+import com.ss.mqtt.broker.service.ClientIdRegistry;
 import com.ss.mqtt.broker.service.ClientService;
 import com.ss.mqtt.broker.service.PublishingService;
 import com.ss.mqtt.broker.service.SubscriptionService;
-import com.ss.mqtt.broker.service.impl.DefaultClientService;
-import com.ss.mqtt.broker.service.impl.SimplePublishingService;
-import com.ss.mqtt.broker.service.impl.SimpleSubscriptionService;
-import com.ss.mqtt.broker.service.impl.SimpleSubscriptions;
+import com.ss.mqtt.broker.service.impl.*;
 import com.ss.rlib.network.*;
 import com.ss.rlib.network.impl.DefaultBufferAllocator;
 import com.ss.rlib.network.server.ServerNetwork;
@@ -52,29 +52,62 @@ private interface ChannelFactory extends
         return new DefaultClientService();
     }
 
+    @NotNull
+    @Bean ClientIdRegistry clientIdRegistry() {
+        return new SimpleClientIdRegistry(
+            env.getProperty(
+                "client.id.available.chars",
+                "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ-"
+            ),
+            env.getProperty("client.id.max.length", int.class, 36)
+        );
+    }
+
     @Bean
-    @NotNull Network<? extends Connection<MqttReadablePacket, MqttWritablePacket>> network(
-        @NotNull ServerNetworkConfig networkConfig,
-        @NotNull BufferAllocator bufferAllocator,
-        @NotNull Consumer<MqttConnection> mqttConnectionConsumer,
-        @NotNull MqttConnectionConfig connectionConfig,
+    PacketInHandler @NotNull [] devicePacketHandlers(
+        @NotNull ClientIdRegistry clientIdRegistry,
         @NotNull SubscriptionService subscriptionService,
         @NotNull PublishingService publishingService
     ) {
-        ServerNetwork<MqttConnection> serverNetwork = NetworkFactory.newServerNetwork(
+
+        var handlers = new PacketInHandler[PacketType.INVALID.ordinal()];
+        handlers[PacketType.CONNECT.ordinal()] = new ConnectInPacketHandler(clientIdRegistry);
+        handlers[PacketType.SUBSCRIBE.ordinal()] = new SubscribeInPacketHandler(subscriptionService);
+        handlers[PacketType.UNSUBSCRIBE.ordinal()] = new UnsubscribeInPacketHandler(subscriptionService);
+        handlers[PacketType.PUBLISH.ordinal()] = new PublishInPacketHandler(publishingService);
+
+        return handlers;
+    }
+
+    @Bean
+    @NotNull ServerNetwork<@NotNull MqttConnection> deviceNetwork(
+        @NotNull ServerNetworkConfig networkConfig,
+        @NotNull BufferAllocator bufferAllocator,
+        @NotNull MqttConnectionConfig deviceConnectionConfig,
+        PacketInHandler @NotNull [] devicePacketHandlers
+    ) {
+        return NetworkFactory.newServerNetwork(
             networkConfig,
-            networkChannelFactory(
+            deviceConnectionFactory(
                 bufferAllocator,
-                connectionConfig,
-                subscriptionService,
-                publishingService
+                deviceConnectionConfig,
+                devicePacketHandlers
             )
         );
+    }
 
-        serverNetwork.start(new InetSocketAddress("localhost", 1883));
-        serverNetwork.onAccept(mqttConnectionConsumer);
+    @Bean
+    @NotNull InetSocketAddress deviceNetworkAddress(
+        @NotNull ServerNetwork<@NotNull MqttConnection> deviceNetwork,
+        @NotNull Consumer<@NotNull MqttConnection> mqttConnectionConsumer
+    ) {
+
+        var address = new InetSocketAddress("localhost", 1883);
+
+        deviceNetwork.start(address);
+        deviceNetwork.onAccept(mqttConnectionConsumer);
 
-        return serverNetwork;
+        return address;
     }
 
     @Bean
@@ -88,16 +121,16 @@ private interface ChannelFactory extends
     }
 
     @Bean
-    @NotNull Consumer<MqttConnection> mqttConnectionConsumer(@NotNull ClientService clientService) {
+    @NotNull Consumer<@NotNull MqttConnection> mqttConnectionConsumer() {
         return mqttConnection -> {
             log.info("Accepted connection: {}", mqttConnection);
-            var client = mqttConnection.getClient();
+            var client = (UnsafeMqttClient) mqttConnection.getClient();
             mqttConnection.onReceive((conn, packet) -> client.handle(packet));
         };
     }
 
     @Bean
-    @NotNull MqttConnectionConfig mqttConnectionConfig() {
+    @NotNull MqttConnectionConfig deviceConnectionConfig() {
         return new MqttConnectionConfig(
             QoS.of(env.getProperty("mqtt.connection.max.qos", int.class, 2)),
             env.getProperty(
@@ -128,21 +161,19 @@ private interface ChannelFactory extends
         );
     }
 
-    private @NotNull ChannelFactory networkChannelFactory(
+    private @NotNull ChannelFactory deviceConnectionFactory(
         @NotNull BufferAllocator bufferAllocator,
         @NotNull MqttConnectionConfig connectionConfig,
-        @NotNull SubscriptionService subscriptionService,
-        @NotNull PublishingService publishingService
+        PacketInHandler @NotNull [] packetHandlers
     ) {
         return (network, channel) -> new MqttConnection(
             network,
             channel,
-            NetworkCryptor.NULL,
             bufferAllocator,
             100,
-            subscriptionService,
-            publishingService,
-            connectionConfig
+            packetHandlers,
+            connectionConfig,
+            DeviceMqttClient::new
         );
     }
 }

src/main/java/com/ss/mqtt/broker/exception/MalformedPacketMqttException.java

@@ -0,0 +1,3 @@
+package com.ss.mqtt.broker.exception;
+
+public class MalformedPacketMqttException extends MqttException {}

src/main/java/com/ss/mqtt/broker/network/MqttClient.java

@@ -2,12 +2,13 @@
 
 import com.ss.mqtt.broker.config.MqttConnectionConfig;
 import com.ss.mqtt.broker.model.MqttVersion;
+import com.ss.mqtt.broker.network.client.MqttClient;
+import com.ss.mqtt.broker.network.client.UnsafeMqttClient;
 import com.ss.mqtt.broker.network.packet.MqttPacketReader;
 import com.ss.mqtt.broker.network.packet.MqttPacketWriter;
 import com.ss.mqtt.broker.network.packet.in.MqttReadablePacket;
+import com.ss.mqtt.broker.network.packet.in.handler.PacketInHandler;
 import com.ss.mqtt.broker.network.packet.out.MqttWritablePacket;
-import com.ss.mqtt.broker.service.PublishingService;
-import com.ss.mqtt.broker.service.SubscriptionService;
 import com.ss.rlib.network.BufferAllocator;
 import com.ss.rlib.network.Connection;
 import com.ss.rlib.network.Network;
@@ -22,40 +23,40 @@
 import org.jetbrains.annotations.NotNull;
 
 import java.nio.channels.AsynchronousSocketChannel;
+import java.util.function.Function;
 
 @Log4j2
-@Getter(AccessLevel.PROTECTED)
 public class MqttConnection extends AbstractConnection<MqttReadablePacket, MqttWritablePacket> {
 
-    private final PacketReader packetReader;
-    private final PacketWriter packetWriter;
+    @Getter(AccessLevel.PROTECTED)
+    private final @NotNull PacketReader packetReader;
 
-    private final SubscriptionService subscriptionService;
-    private final PublishingService publishingService;
+    @Getter(AccessLevel.PROTECTED)
+    private final @NotNull PacketWriter packetWriter;
+
+    private final @Getter PacketInHandler @NotNull [] packetHandlers;
 
     private final @Getter @NotNull MqttClient client;
     private final @Getter @NotNull MqttConnectionConfig config;
 
-    private volatile @Setter @NotNull MqttVersion mqttVersion;
+    private volatile @Getter @Setter @NotNull MqttVersion mqttVersion;
 
     public MqttConnection(
         @NotNull Network<? extends Connection<MqttReadablePacket, MqttWritablePacket>> network,
         @NotNull AsynchronousSocketChannel channel,
-        @NotNull NetworkCryptor crypt,
         @NotNull BufferAllocator bufferAllocator,
         int maxPacketsByRead,
-        @NotNull SubscriptionService subscriptionService,
-        @NotNull PublishingService publishingService,
-        @NotNull MqttConnectionConfig config
+        PacketInHandler @NotNull [] packetHandlers,
+        @NotNull MqttConnectionConfig config,
+        @NotNull Function<MqttConnection, UnsafeMqttClient> clientFactory
     ) {
-        super(network, channel, crypt, bufferAllocator, maxPacketsByRead);
+        super(network, channel, NetworkCryptor.NULL, bufferAllocator, maxPacketsByRead);
+        this.packetHandlers = packetHandlers;
         this.config = config;
         this.mqttVersion = MqttVersion.MQTT_3_1_1;
         this.packetReader = createPacketReader();
         this.packetWriter = createPacketWriter();
-        this.client = new MqttClient(this);
-        this.subscriptionService = subscriptionService;
-        this.publishingService = publishingService;
+        this.client = clientFactory.apply(this);
     }
 
     public boolean isSupported(@NotNull MqttVersion mqttVersion) {

src/main/java/com/ss/mqtt/broker/network/MqttConnection.java

@@ -0,0 +1,22 @@
+package com.ss.mqtt.broker.network.client;
+
+import com.ss.mqtt.broker.config.MqttConnectionConfig;
+import com.ss.mqtt.broker.network.packet.factory.MqttPacketOutFactory;
+import com.ss.mqtt.broker.network.packet.out.MqttWritablePacket;
+import org.jetbrains.annotations.NotNull;
+
+public interface MqttClient {
+
+    @NotNull MqttPacketOutFactory getPacketOutFactory();
+    @NotNull MqttConnectionConfig getConnectionConfig();
+    @NotNull String getClientId();
+
+    int getKeepAlive();
+    int getMaximumPacketSize();
+    int getReceiveMax();
+    int getTopicAliasMaximum();
+
+    long getSessionExpiryInterval();
+
+    void send(@NotNull MqttWritablePacket packet);
+}