continue updating network API to support MQTT implementation

Author: JavaSaBr

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

@@ -55,7 +55,6 @@ public abstract class AbstractConnection<R extends ReadablePacket, W extends Wri
     protected final Array<BiConsumer<? super Connection<R, W>, ? super R>> subscribers;
 
     protected final int maxPacketsByRead;
-    protected final int packetLengthHeaderSize;
 
     @Getter
     protected volatile long lastActivity;
@@ -65,12 +64,10 @@ public AbstractConnection(
         @NotNull AsynchronousSocketChannel channel,
         @NotNull NetworkCryptor crypt,
         @NotNull BufferAllocator bufferAllocator,
-        int maxPacketsByRead,
-        int packetLengthHeaderSize
+        int maxPacketsByRead
     ) {
         this.bufferAllocator = bufferAllocator;
         this.maxPacketsByRead = maxPacketsByRead;
-        this.packetLengthHeaderSize = packetLengthHeaderSize;
         this.lock = new StampedLock();
         this.crypt = crypt;
         this.channel = channel;

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

@@ -19,15 +19,15 @@
 /**
  * @author JavaSaBr
  */
+@Getter(AccessLevel.PROTECTED)
 public abstract class DefaultDataConnection<R extends ReadablePacket, W extends WritablePacket> extends
     AbstractConnection<R, W> {
 
-    @Getter(AccessLevel.PROTECTED)
     private final PacketReader packetReader;
-
-    @Getter(AccessLevel.PROTECTED)
     private final PacketWriter packetWriter;
 
+    private final int packetLengthHeaderSize;
+
     public DefaultDataConnection(
         @NotNull Network<? extends Connection<R, W>> network,
         @NotNull AsynchronousSocketChannel channel,
@@ -36,7 +36,8 @@ public DefaultDataConnection(
         int maxPacketsByRead,
         int packetLengthHeaderSize
     ) {
-        super(network, channel, crypt, bufferAllocator, maxPacketsByRead, packetLengthHeaderSize);
+        super(network, channel, crypt, bufferAllocator, maxPacketsByRead);
+        this.packetLengthHeaderSize = packetLengthHeaderSize;
         this.packetReader = createPacketReader();
         this.packetWriter = createPacketWriter();
     }

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

@@ -28,6 +28,7 @@ public class IdBasedPacketConnection<R extends IdBasedReadablePacket<R>, W exten
     private final PacketWriter packetWriter;
     private final ReadablePacketRegistry<R> packetRegistry;
 
+    private final int packetLengthHeaderSize;
     private final int packetIdHeaderSize;
 
     public IdBasedPacketConnection(
@@ -40,8 +41,9 @@ public IdBasedPacketConnection(
         int packetLengthHeaderSize,
         int packetIdHeaderSize
     ) {
-        super(network, channel, crypt, bufferAllocator, maxPacketsByRead, packetLengthHeaderSize);
+        super(network, channel, crypt, bufferAllocator, maxPacketsByRead);
         this.packetRegistry = packetRegistry;
+        this.packetLengthHeaderSize = packetLengthHeaderSize;
         this.packetIdHeaderSize = packetIdHeaderSize;
         this.packetReader = createPacketReader();
         this.packetWriter = createPacketWriter();

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

@@ -245,9 +245,9 @@ else if (packetLength > readTempBuffer.capacity()) {
             R packet;
 
             if (decryptedData != null) {
-                packet = createPacketFor(decryptedData, packetLength, dataLength);
+                packet = createPacketFor(decryptedData, positionBeforeRead, packetLength, dataLength);
             } else {
-                packet = createPacketFor(bufferToRead, packetLength, dataLength);
+                packet = createPacketFor(bufferToRead, positionBeforeRead, packetLength, dataLength);
             }
 
             if (packet != null) {
@@ -259,6 +259,8 @@ else if (packetLength > readTempBuffer.capacity()) {
             } else {
                 LOGGER.warning("Cannot create any instance of packet to read data.");
             }
+
+            bufferToRead.position(endPosition);
         }
 
         if (bufferToRead.hasRemaining()) {
@@ -298,7 +300,9 @@ else if (packetLength > readTempBuffer.capacity()) {
      * @param buffer       the buffer.
      * @return the length of packet data part.
      */
-    protected abstract int calcDataLength(int packetLength, int readBytes, @NotNull ByteBuffer buffer);
+    protected int calcDataLength(int packetLength, int readBytes, @NotNull ByteBuffer buffer) {
+        return packetLength - readBytes;
+    }
 
     /**
      * Get the packet's data length of next packet in the buffer.
@@ -423,12 +427,18 @@ protected int readHeader(@NotNull ByteBuffer buffer, int headerSize) {
     /**
      * Create a packet to read received data.
      *
-     * @param buffer       the buffer with received data.
-     * @param packetLength the length of packet.
-     * @param dataLength   length of packet's data.
+     * @param buffer              the buffer with received data.
+     * @param startPacketPosition the start position of the packet in the buffer.
+     * @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 packetLength, int dataLength);
+    protected abstract @Nullable R createPacketFor(
+        @NotNull ByteBuffer buffer,
+        int startPacketPosition,
+        int packetLength,
+        int dataLength
+    );
 
     @Override
     public void close() {

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

@@ -49,18 +49,18 @@ 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) {
+    protected @Nullable R createPacketFor(
+        @NotNull ByteBuffer buffer,
+        int startPacketPosition,
+        int packetLength,
+        int dataLength
+    ) {
         return readPacketFactory.apply(dataLength);
     }
 }

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

@@ -52,18 +52,19 @@ 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();
+    protected @Nullable R createPacketFor(
+        @NotNull ByteBuffer buffer,
+        int startPacketPosition,
+        int packetLength,
+        int dataLength
+    ) {
+        return packetRegistry.findById(readHeader(buffer, packetIdHeaderSize))
+            .newInstance();
     }
 }