improving network module

Author: JavaSaBr

README.md

@@ -14,7 +14,7 @@ repositories {
 }
 
 ext {
-   rlibVersion = "10.0.alpha"
+   rlibVersion = "10.0.alpha3"
 }
 
 dependencies {

rlib-network/src/loadTest/java/javasabr/rlib/network/StringNetworkLoadTest.java

@@ -19,7 +19,6 @@
 import javasabr.rlib.network.client.ClientNetwork;
 import javasabr.rlib.network.impl.DefaultBufferAllocator;
 import javasabr.rlib.network.impl.StringDataConnection;
-import javasabr.rlib.network.impl.StringDataSslConnection;
 import javasabr.rlib.network.packet.impl.StringReadableNetworkPacket;
 import javasabr.rlib.network.packet.impl.StringWritableNetworkPacket;
 import javasabr.rlib.network.server.ServerNetwork;

rlib-network/src/loadTest/java/javasabr/rlib/network/StringSslNetworkLoadTest.java

@@ -19,10 +19,7 @@
 import javasabr.rlib.network.ServerNetworkConfig.SimpleServerNetworkConfig;
 import javasabr.rlib.network.client.ClientNetwork;
 import javasabr.rlib.network.impl.DefaultBufferAllocator;
-import javasabr.rlib.network.impl.StringDataConnection;
 import javasabr.rlib.network.impl.StringDataSslConnection;
-import javasabr.rlib.network.packet.impl.AbstractSslNetworkPacketReader;
-import javasabr.rlib.network.packet.impl.AbstractSslNetworkPacketWriter;
 import javasabr.rlib.network.packet.impl.StringReadableNetworkPacket;
 import javasabr.rlib.network.packet.impl.StringWritableNetworkPacket;
 import javasabr.rlib.network.server.ServerNetwork;

rlib-network/src/main/java/javasabr/rlib/network/BufferAllocator.java

@@ -1,7 +1,6 @@
 package javasabr.rlib.network;
 
 import java.nio.ByteBuffer;
-import org.jspecify.annotations.Nullable;
 
 /**
  * The interface to implement a buffer allocator for network things.

rlib-network/src/main/java/javasabr/rlib/network/NetworkConfig.java

@@ -83,6 +83,13 @@ default int retryDelayInMs() {
     return 1000;
   }
 
+  /**
+   * Gets a max allowed empty reads from socket channel before closing a connection.
+   */
+  default int maxEmptyReadsBeforeClose() {
+    return 3;
+  }
+
   default ByteOrder byteOrder() {
     return ByteOrder.BIG_ENDIAN;
   }

rlib-network/src/main/java/javasabr/rlib/network/impl/StringDataSslConnection.java

@@ -4,7 +4,6 @@
 import javasabr.rlib.network.BufferAllocator;
 import javasabr.rlib.network.Network;
 import javasabr.rlib.network.packet.impl.StringReadableNetworkPacket;
-import javasabr.rlib.network.packet.impl.StringWritableNetworkPacket;
 import javax.net.ssl.SSLContext;
 
 /**

rlib-network/src/main/java/javasabr/rlib/network/packet/impl/AbstractNetworkPacketReader.java

@@ -5,15 +5,15 @@
 import java.nio.ByteBuffer;
 import java.nio.channels.AsynchronousCloseException;
 import java.nio.channels.AsynchronousSocketChannel;
+import java.nio.channels.ClosedChannelException;
 import java.nio.channels.CompletionHandler;
 import java.nio.channels.InterruptedByTimeoutException;
-import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Consumer;
 import javasabr.rlib.common.util.BufferUtils;
 import javasabr.rlib.network.BufferAllocator;
-import javasabr.rlib.network.Connection;
 import javasabr.rlib.network.Network;
 import javasabr.rlib.network.NetworkConfig;
 import javasabr.rlib.network.UnsafeConnection;
@@ -55,6 +55,7 @@ public void failed(Throwable exc, ByteBuffer readingBuffer) {
   };
 
   final AtomicBoolean reading = new AtomicBoolean(false);
+  final AtomicInteger emptyReadsCounter = new AtomicInteger(0);
 
   final C connection;
   final AsynchronousSocketChannel socketChannel;
@@ -117,8 +118,9 @@ protected void startReadImpl() {
       socketChannel.read(buffer, buffer, readChannelHandler);
     } catch (RuntimeException ex) {
       log.error(ex);
-      reading.compareAndSet(true, false);
-      retryReadLater();
+      if (reading.compareAndSet(true, false)) {
+        retryReadLater();
+      }
     }
   }
 
@@ -392,20 +394,11 @@ protected void freeTempBigBuffers() {
   protected void handleReceivedData(int receivedBytes, ByteBuffer readingBuffer) {
     updateActivityFunction.run();
     if (receivedBytes == -1) {
-      log.debug(remoteAddress(), "[%s] Received empty bytes from channel"::formatted);
-      if (connection.closed()) {
-        reading.compareAndSet(true, false);
-        return;
-      } else if (!socketChannel.isOpen()) {
-        connection.close();
-        return;
-      }
-      if (reading.compareAndSet(false, true)) {
-        retryReadLater();
-      }
+      handleEmptyReadFromChannel();
     } else {
       log.debug(remoteAddress(), receivedBytes, "[%s] Received [%s] bytes from channel"::formatted);
       readingBuffer.flip();
+      emptyReadsCounter.set(0);
       try {
         readPackets(readingBuffer);
       } catch (Exception e) {
@@ -415,6 +408,31 @@ protected void handleReceivedData(int receivedBytes, ByteBuffer readingBuffer) {
     }
   }
 
+  protected void handleEmptyReadFromChannel() {
+    log.debug(remoteAddress(), "[%s] Received empty bytes from channel"::formatted);
+
+    if (connection.closed()) {
+      reading.compareAndSet(true, false);
+      return;
+    } else if (!socketChannel.isOpen()) {
+      connection.close();
+      return;
+    }
+
+    NetworkConfig config = connection
+        .network()
+        .config();
+
+    if (emptyReadsCounter.incrementAndGet() > config.maxEmptyReadsBeforeClose()) {
+      connection.close();
+      return;
+    }
+
+    if (reading.compareAndSet(true, false)) {
+      retryReadLater();
+    }
+  }
+
   /**
    * Handles the exception during receiving data from the channel.
    *
@@ -424,12 +442,14 @@ protected void handleReceivedData(int receivedBytes, ByteBuffer readingBuffer) {
   protected void handleFailedReceiving(Throwable exception, ByteBuffer readingBuffer) {
     if (exception instanceof InterruptedByTimeoutException) {
       if (reading.compareAndSet(true, false)) {
-        startRead();
+        retryReadLater();
       }
       return;
     }
     if (exception instanceof AsynchronousCloseException) {
       log.info(remoteAddress(), "[%s] Connection was closed"::formatted);
+    } else if (exception instanceof ClosedChannelException) {
+      log.info(remoteAddress(), "[%s] Connection was closed"::formatted);
     } else {
       log.error(exception);
       connection.close();

rlib-network/src/main/java/javasabr/rlib/network/packet/impl/AbstractReusableWritableNetworkPacket.java

@@ -9,7 +9,6 @@
 import javasabr.rlib.common.util.ClassUtils;
 import javasabr.rlib.network.Connection;
 import javasabr.rlib.network.packet.ReusableWritablePacket;
-import javasabr.rlib.network.packet.WritableNetworkPacket;
 import javasabr.rlib.reusable.pool.Pool;
 import javasabr.rlib.reusable.pool.PoolFactory;
 import lombok.AccessLevel;

rlib-network/src/main/java/javasabr/rlib/network/packet/impl/IdBasedNetworkPacketWriter.java

@@ -4,9 +4,6 @@
 import java.nio.channels.AsynchronousSocketChannel;
 import java.util.function.Consumer;
 import java.util.function.Supplier;
-import javasabr.rlib.common.function.NotNullBiConsumer;
-import javasabr.rlib.common.function.NotNullConsumer;
-import javasabr.rlib.common.function.NullableSupplier;
 import javasabr.rlib.functions.ObjBoolConsumer;
 import javasabr.rlib.network.BufferAllocator;
 import javasabr.rlib.network.Connection;