add test

Author: JavaSaBr

buildSrc/src/main/groovy/configure-java.gradle

@@ -35,6 +35,7 @@ javadoc {
 test {
   useJUnitPlatform()
   failOnNoDiscoveredTests = false
+  jvmArgs("-XX:+EnableDynamicAgentLoading")
 }
 
 tasks.register("loadTest", Test) {
@@ -52,6 +53,8 @@ dependencies {
   compileOnly libs.lombok
   annotationProcessor libs.lombok
 
+  testImplementation libs.mockito.core
+  testImplementation libs.mockito.junit.jupiter
   testImplementation libs.junit.api
   testImplementation libs.junit.jupiter.params
 

gradle/libs.versions.toml

@@ -17,6 +17,8 @@ lombok = "1.18.38"
 jspecify = "1.0.0"
 # https://mvnrepository.com/artifact/org.junit.platform/junit-platform-launcher
 junit-platform-launcher = "1.13.4"
+# https://mvnrepository.com/artifact/org.mockito/mockito-core
+mockito = "5.20.0"
 
 [libraries]
 project-reactor-core = { module = "io.projectreactor:reactor-core", version.ref = "project-reactor" }
@@ -32,6 +34,8 @@ slf4j-ext = { module = "org.slf4j:slf4j-ext", version.ref = "slf4j" }
 jakarta-mail-api = { module = "jakarta.mail:jakarta.mail-api", version.ref = "jakarta-mail" }
 angus-mail = { module = "org.eclipse.angus:angus-mail", version.ref = "angus-mail" }
 testcontainers = { module = "org.testcontainers:testcontainers", version.ref = "testcontainers" }
+mockito-core = { module = "org.mockito:mockito-core", version.ref = "mockito" }
+mockito-junit-jupiter = { module = "org.mockito:mockito-junit-jupiter", version.ref = "mockito" }
 
 [bundles]
 mail = ["jakarta-mail-api", "angus-mail"]

rlib-network/src/main/java/javasabr/rlib/network/exception/MalformedProtocolException.java

@@ -0,0 +1,7 @@
+package javasabr.rlib.network.exception;
+
+public class MalformedProtocolException extends RuntimeException {
+  public MalformedProtocolException(String message) {
+    super(message);
+  }
+}

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

@@ -17,6 +17,7 @@
 import javasabr.rlib.network.Network;
 import javasabr.rlib.network.NetworkConfig;
 import javasabr.rlib.network.UnsafeConnection;
+import javasabr.rlib.network.exception.MalformedProtocolException;
 import javasabr.rlib.network.packet.NetworkPacketReader;
 import javasabr.rlib.network.packet.ReadableNetworkPacket;
 import lombok.AccessLevel;
@@ -121,8 +122,6 @@ protected void startReadImpl() {
       if (reading.compareAndSet(true, false)) {
         retryReadLater();
       }
-    } catch (Error error) {
-      throw error;
     }
   }
 
@@ -219,7 +218,7 @@ else if (waitedBytes > 0) {
       int positionBeforeRead = endPosition;
       int packetFullLength = readFullPacketLength(bufferToRead);
       if (packetFullLength > networkConfig.maxPacketSize()) {
-        throw new IllegalStateException(
+        throw new MalformedProtocolException(
             "Received to big packet:[" + packetFullLength + ">" + networkConfig.maxPacketSize() + "]");
       }
 
@@ -412,8 +411,9 @@ protected void handleReceivedData(int receivedBytes, ByteBuffer readingBuffer) {
       emptyReadsCounter.set(0);
       try {
         readPackets(readingBuffer);
-      } catch (Exception e) {
-        log.error(e);
+      } catch (MalformedProtocolException e) {
+        handleFailedReceiving(e, readingBuffer);
+        return;
       }
       startReadImpl();
     }
@@ -451,19 +451,20 @@ protected void handleEmptyReadFromChannel() {
    * @param readingBuffer the currently reading buffer.
    */
   protected void handleFailedReceiving(Throwable exception, ByteBuffer readingBuffer) {
-    if (exception instanceof InterruptedByTimeoutException) {
-      if (reading.compareAndSet(true, false)) {
-        retryReadLater();
+    switch (exception) {
+      case InterruptedByTimeoutException ex -> {
+        if (reading.compareAndSet(true, false)) {
+          retryReadLater();
+        }
+      }
+      case AsynchronousCloseException ex ->
+          log.info(remoteAddress(), "[%s] Connection was closed"::formatted);
+      case ClosedChannelException ex ->
+          log.info(remoteAddress(), "[%s] Connection was closed"::formatted);
+      default -> {
+        log.error(exception);
+        connection.close();
       }
-      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();
     }
   }
 
@@ -475,16 +476,12 @@ protected int maxPacketsByRead() {
   }
 
   protected int readHeader(ByteBuffer buffer, int headerSize) {
-    switch (headerSize) {
-      case 1:
-        return buffer.get() & 0xFF;
-      case 2:
-        return buffer.getShort() & 0xFFFF;
-      case 4:
-        return buffer.getInt();
-      default:
-        throw new IllegalStateException("Wrong packet's header size: " + headerSize);
-    }
+    return switch (headerSize) {
+      case 1 -> buffer.get() & 0xFF;
+      case 2 -> buffer.getShort() & 0xFFFF;
+      case 4 -> buffer.getInt();
+      default -> throw new MalformedProtocolException("Wrong packet's header size:" + headerSize);
+    };
   }
 
   /**

rlib-network/src/test/java/javasabr/rlib/network/impl/DefaultDataConnectionTest.java

@@ -0,0 +1,88 @@
+package javasabr.rlib.network.impl;
+
+import java.nio.ByteBuffer;
+import java.nio.channels.AsynchronousSocketChannel;
+import java.nio.channels.CompletionHandler;
+import java.nio.charset.StandardCharsets;
+import javasabr.rlib.network.BufferAllocator;
+import javasabr.rlib.network.Network;
+import javasabr.rlib.network.NetworkConfig;
+import javasabr.rlib.network.ServerNetworkConfig.SimpleServerNetworkConfig;
+import javasabr.rlib.network.packet.ReadableNetworkPacket;
+import javasabr.rlib.network.packet.impl.DefaultNetworkPacketReader;
+import org.jspecify.annotations.NonNull;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.Mockito;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+@ExtendWith(MockitoExtension.class)
+class DefaultDataConnectionTest {
+
+  private static class TestDataConnection extends DefaultDataConnection<TestDataConnection> {
+
+    public TestDataConnection(
+        Network<TestDataConnection> network,
+        AsynchronousSocketChannel channel,
+        BufferAllocator bufferAllocator,
+        int maxPacketsByRead,
+        int packetLengthHeaderSize) {
+      super(network, channel, bufferAllocator, maxPacketsByRead, packetLengthHeaderSize);
+    }
+
+    @Override
+    protected ReadableNetworkPacket<TestDataConnection> createReadablePacket() {
+      throw new UnsupportedOperationException();
+    }
+  }
+
+  @Mock
+  Network<@NonNull TestDataConnection> network;
+
+  @Mock
+  AsynchronousSocketChannel channel;
+
+  @Test
+  @DisplayName("should not allow to read too big packet")
+  void shouldNotAllowToReadTooBigPacket() {
+    // given:
+    NetworkConfig networkConfig = SimpleServerNetworkConfig
+        .builder()
+        .build();
+
+    var bufferAllocator = new DefaultBufferAllocator(networkConfig);
+    var connection = new TestDataConnection(network, channel, bufferAllocator, 100, 4);
+    var packetReader = (DefaultNetworkPacketReader<?, ?>) connection.packetReader;
+
+    int nextPacketSize = 250_000_000;
+
+    var incomingBuffer = ByteBuffer.allocate(100);
+    incomingBuffer.putInt(nextPacketSize);
+    incomingBuffer.put("Test data".getBytes(StandardCharsets.UTF_8));
+
+    Mockito
+        .when(network.config())
+        .thenReturn(networkConfig);
+
+    Mockito
+        .doAnswer(invocationOnMock -> {
+          CompletionHandler<Integer, ByteBuffer> readChannelHandler = invocationOnMock.getArgument(2);
+          readChannelHandler.completed(incomingBuffer.remaining(), incomingBuffer);
+          return null;
+        })
+        .when(channel)
+        .read(Mockito.any(), Mockito.any(), Mockito.any());
+
+    // then:
+    Assertions.assertFalse(connection.closed());
+
+    // when:
+    packetReader.startRead();
+
+    // then:
+    Assertions.assertTrue(connection.closed());
+  }
+}