update network API to support MQTT implementation

Author: JavaSaBr

rlib-common/src/main/java/com/ss/rlib/common/concurrent/GroupThreadFactory.java

@@ -20,30 +20,13 @@ public interface ThreadConstructor {
         @NotNull Thread create(@NotNull ThreadGroup group, @NotNull Runnable runnable, @NotNull String name);
     }
 
-    /**
-     * The order of the next thread.
-     */
     private final AtomicInteger ordinal;
-
-    /**
-     * The name of this group.
-     */
     private final String name;
-
-    /**
-     * The group of threads.
-     */
     private final ThreadGroup group;
-
-    /**
-     * The thread's constructor.
-     */
     private final ThreadConstructor constructor;
 
-    /**
-     * The priority of these threads.
-     */
     private final int priority;
+    private final boolean daemon;
 
     public GroupThreadFactory(@NotNull String name) {
         this(name, Thread::new, Thread.NORM_PRIORITY);
@@ -54,11 +37,21 @@ public GroupThreadFactory(@NotNull String name, int priority) {
     }
 
     public GroupThreadFactory(@NotNull String name, @NotNull ThreadConstructor constructor, int priority) {
+        this(name, constructor, priority, false);
+    }
+
+    public GroupThreadFactory(
+        @NotNull String name,
+        @NotNull ThreadConstructor constructor,
+        int priority,
+        boolean daemon
+    ) {
         this.constructor = constructor;
         this.priority = priority;
         this.name = name;
         this.group = new ThreadGroup(name);
         this.ordinal = new AtomicInteger();
+        this.daemon = daemon;
     }
 
     @Deprecated
@@ -67,6 +60,7 @@ public GroupThreadFactory(@NotNull String name, @NotNull Class<? extends Thread>
         this.name = name;
         this.group = new ThreadGroup(name);
         this.ordinal = new AtomicInteger();
+        this.daemon = false;
         this.constructor = new ThreadConstructor() {
 
             Constructor<? extends Thread> constructor =
@@ -87,6 +81,7 @@ public GroupThreadFactory(@NotNull String name, @NotNull Class<? extends Thread>
     public @NotNull Thread newThread(@NotNull Runnable runnable) {
         var thread = constructor.create(group, runnable, name + "-" + ordinal.incrementAndGet());
         thread.setPriority(priority);
+        thread.setDaemon(daemon);
         return thread;
     }
 }

rlib-network/src/main/java/com/ss/rlib/network/impl/IdBasedPacketConnection.java

@@ -20,16 +20,14 @@
 /**
  * @author JavaSaBr
  */
+@Getter(AccessLevel.PROTECTED)
 public class IdBasedPacketConnection<R extends IdBasedReadablePacket<R>, W extends IdBasedWritablePacket> extends
     AbstractConnection<R, W> {
 
-    @Getter(AccessLevel.PROTECTED)
     private final PacketReader packetReader;
-
-    @Getter(AccessLevel.PROTECTED)
     private final PacketWriter packetWriter;
-
     private final ReadablePacketRegistry<R> packetRegistry;
+
     private final int packetIdHeaderSize;
 
     public IdBasedPacketConnection(

rlib-network/src/main/java/com/ss/rlib/network/packet/impl/AbstractPacketReader.java

@@ -64,7 +64,6 @@ public void failed(@NotNull Throwable exc, @Nullable Void attachment) {
     @Setter(AccessLevel.PROTECTED)
     protected volatile ByteBuffer decryptedTempBuffer;
 
-    protected final int packetLengthHeaderSize;
     protected final int maxPacketsByRead;
 
     protected AbstractPacketReader(
@@ -73,7 +72,6 @@ protected AbstractPacketReader(
         @NotNull BufferAllocator bufferAllocator,
         @NotNull Runnable updateActivityFunction,
         @NotNull Consumer<? super R> readPacketHandler,
-        int packetLengthHeaderSize,
         int maxPacketsByRead
     ) {
         this.connection = connection;
@@ -84,7 +82,6 @@ protected AbstractPacketReader(
         this.decryptedBuffer = bufferAllocator.takePendingBuffer();
         this.updateActivityFunction = updateActivityFunction;
         this.readPacketHandler = readPacketHandler;
-        this.packetLengthHeaderSize = packetLengthHeaderSize;
         this.maxPacketsByRead = maxPacketsByRead;
     }
 
@@ -165,24 +162,25 @@ else if (waitedBytes > 0) {
         var readPackets = 0;
         var endPosition = 0;
 
-        while (bufferToRead.remaining() >= packetLengthHeaderSize && readPackets < maxPacketsByRead) {
+        while (canStartReadPacket(bufferToRead) && readPackets < maxPacketsByRead) {
 
             // set position of start a next packet
             bufferToRead.position(endPosition);
 
+            var positionBeforeRead = endPosition;
             var packetLength = readPacketLength(bufferToRead);
-            var dataLength = packetLength - packetLengthHeaderSize;
+            var dataLength = calcDataLength(packetLength, bufferToRead.position() - endPosition, bufferToRead);
 
-            LOGGER.debug(packetLength, bufferToRead.position() - packetLengthHeaderSize,
+            LOGGER.debug(packetLength, positionBeforeRead,
                 (length, pos) -> "Find next packet from position: " + pos + " with length: " + length);
 
             // calculate position of end the next packet
             endPosition += packetLength;
 
             // if the packet isn't full presented in this buffer
-            if (endPosition > bufferToRead.limit()) {
+            if (packetLength == -1 || endPosition > bufferToRead.limit()) {
 
-                bufferToRead.position(bufferToRead.position() - packetLengthHeaderSize);
+                bufferToRead.position(positionBeforeRead);
 
                 // if we read the received buffer we need to put
                 // not read data to the pending buffer or big mapped byte buffer
@@ -247,9 +245,9 @@ else if (packetLength > readTempBuffer.capacity()) {
             R packet;
 
             if (decryptedData != null) {
-                packet = createPacketFor(decryptedData, dataLength);
+                packet = createPacketFor(decryptedData, packetLength, dataLength);
             } else {
-                packet = createPacketFor(bufferToRead, dataLength);
+                packet = createPacketFor(bufferToRead, packetLength, dataLength);
             }
 
             if (packet != null) {
@@ -284,6 +282,32 @@ else if (packetLength > readTempBuffer.capacity()) {
         return readPackets;
     }
 
+    /**
+     * Check buffer's data.
+     *
+     * @param buffer the buffer to read.
+     * @return true if this buffer has enough data to start initial reading.
+     */
+    protected abstract boolean canStartReadPacket(@NotNull ByteBuffer buffer);
+
+    /**
+     * Calculate size of packet data.
+     *
+     * @param packetLength the full packet length.
+     * @param readBytes    the count of already read bytes from buffer to get packet length.
+     * @param buffer       the buffer.
+     * @return the length of packet data part.
+     */
+    protected abstract int calcDataLength(int packetLength, int readBytes, @NotNull ByteBuffer buffer);
+
+    /**
+     * Get the packet's data length of next packet in the buffer.
+     *
+     * @param buffer the buffer with received data.
+     * @return the packet length or -1 if we have no enough data to read length.
+     */
+    protected abstract int readPacketLength(@NotNull ByteBuffer buffer);
+
     protected void reAllocTempBuffers(
         @NotNull ByteBuffer sourceBuffer,
         int packetLength
@@ -383,16 +407,6 @@ protected int getMaxPacketsByRead() {
         return maxPacketsByRead;
     }
 
-    /**
-     * Get the packet's data length of next packet in the buffer.
-     *
-     * @param buffer the buffer with received data.
-     * @return the packet length.
-     */
-    protected int readPacketLength(@NotNull ByteBuffer buffer) {
-        return readHeader(buffer, packetLengthHeaderSize);
-    }
-
     protected int readHeader(@NotNull ByteBuffer buffer, int headerSize) {
         switch (headerSize) {
             case 1:
@@ -409,11 +423,12 @@ protected int readHeader(@NotNull ByteBuffer buffer, int headerSize) {
     /**
      * Create a packet to read received data.
      *
-     * @param buffer the buffer with received data.
-     * @param length length of packet's data.
+     * @param buffer       the buffer with received data.
+     * @param packetLength the length of packet.
+     * @param dataLength   length of packet's data.
      * @return the readable packet.
      */
-    protected abstract @Nullable R createPacketFor(@NotNull ByteBuffer buffer, int length);
+    protected abstract @Nullable R createPacketFor(@NotNull ByteBuffer buffer, int packetLength, int dataLength);
 
     @Override
     public void close() {

rlib-network/src/main/java/com/ss/rlib/network/packet/impl/DefaultPacketReader.java

@@ -20,6 +20,7 @@ public class DefaultPacketReader<R extends ReadablePacket, C extends Connection<
     AbstractPacketReader<R, C> {
 
     private final IntFunction<R> readPacketFactory;
+    private final int packetLengthHeaderSize;
 
     public DefaultPacketReader(
         @NotNull C connection,
@@ -37,14 +38,29 @@ public DefaultPacketReader(
             bufferAllocator,
             updateActivityFunction,
             readPacketHandler,
-            packetLengthHeaderSize,
             maxPacketsByRead
         );
         this.readPacketFactory = readPacketFactory;
+        this.packetLengthHeaderSize = packetLengthHeaderSize;
     }
 
     @Override
-    protected @Nullable R createPacketFor(@NotNull ByteBuffer buffer, int length) {
-        return readPacketFactory.apply(length);
+    protected boolean canStartReadPacket(@NotNull ByteBuffer buffer) {
+        return buffer.remaining() >= packetLengthHeaderSize;
+    }
+
+    @Override
+    protected int calcDataLength(int packetLength, int readBytes, @NotNull ByteBuffer buffer) {
+        return packetLength - packetLengthHeaderSize;
+    }
+
+    @Override
+    protected int readPacketLength(@NotNull ByteBuffer buffer) {
+        return readHeader(buffer, packetLengthHeaderSize);
+    }
+
+    @Override
+    protected @Nullable R createPacketFor(@NotNull ByteBuffer buffer, int packetLength, int dataLength) {
+        return readPacketFactory.apply(dataLength);
     }
 }

rlib-network/src/main/java/com/ss/rlib/network/packet/impl/IdBasedPacketReader.java

@@ -20,6 +20,7 @@ public class IdBasedPacketReader<R extends IdBasedReadablePacket<R>, C extends C
     AbstractPacketReader<R, C> {
 
     private final ReadablePacketRegistry<R> packetRegistry;
+    private final int packetLengthHeaderSize;
     private final int packetIdHeaderSize;
 
     public IdBasedPacketReader(
@@ -39,15 +40,30 @@ public IdBasedPacketReader(
             bufferAllocator,
             updateActivityFunction,
             readPacketHandler,
-            packetLengthHeaderSize,
             maxPacketsByRead
         );
+        this.packetLengthHeaderSize = packetLengthHeaderSize;
         this.packetIdHeaderSize = packetIdHeaderSize;
         this.packetRegistry = packetRegistry;
     }
 
     @Override
-    protected @Nullable R createPacketFor(@NotNull ByteBuffer buffer, int length) {
+    protected boolean canStartReadPacket(@NotNull ByteBuffer buffer) {
+        return buffer.remaining() > packetLengthHeaderSize;
+    }
+
+    @Override
+    protected int calcDataLength(int packetLength, int readBytes, @NotNull ByteBuffer buffer) {
+        return packetLength - packetLengthHeaderSize;
+    }
+
+    @Override
+    protected int readPacketLength(@NotNull ByteBuffer buffer) {
+        return readHeader(buffer, packetLengthHeaderSize);
+    }
+
+    @Override
+    protected @Nullable R createPacketFor(@NotNull ByteBuffer buffer, int packetLength, int dataLength) {
         return packetRegistry.findById(readHeader(buffer, packetIdHeaderSize)).newInstance();
     }
 }

rlib-network/src/main/java/com/ss/rlib/network/server/impl/DefaultServerNetwork.java

@@ -74,7 +74,8 @@ public DefaultServerNetwork(
         var threadFactory = new GroupThreadFactory(
             config.getGroupName(),
             config.getThreadConstructor(),
-            config.getThreadPriority()
+            config.getThreadPriority(),
+            true
         );
 
         var executor = config.getGroupSize() < config.getGroupMaxSize() ? new ThreadPoolExecutor(
@@ -87,6 +88,9 @@ public DefaultServerNetwork(
             new ThreadPoolExecutor.CallerRunsPolicy()
         ) : Executors.newFixedThreadPool(config.getGroupSize(), threadFactory);
 
+        // activate the executor
+        executor.submit(() -> {});
+
         LOGGER.info("Executor configuration:");
         LOGGER.info(config, conf -> "Min threads: " + conf.getGroupSize());
         LOGGER.info(config, conf -> "Max threads: " + conf.getGroupMaxSize());
@@ -123,7 +127,13 @@ public DefaultServerNetwork(
     @Override
     public <S extends ServerNetwork<C>> @NotNull S start(@NotNull InetSocketAddress serverAddress) {
         Utils.unchecked(channel, serverAddress, AsynchronousServerSocketChannel::bind);
+
         LOGGER.info(serverAddress, adr -> "Started server on address: " + adr);
+
+        if (!subscribers.isEmpty()) {
+            acceptNext();
+        }
+
         return ClassUtils.unsafeCast(this);
     }