continue fix ssl network implementation

Author: JavaSaBr

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

@@ -82,7 +82,7 @@ void connectAndSendMessages(
             int delay = random.nextInt(MAX_SEND_DELAY);
             ScheduledFuture<?> schedule = executor.schedule(
                 () -> {
-                  StringWritableNetworkPacket message = newMessage(10, 10240); // 10240
+                  StringWritableNetworkPacket message = newMessage(12000, 15000); // 10240
                   connection.send(message);
                 }, delay, TimeUnit.MILLISECONDS);
             tasks.add(schedule);
@@ -122,7 +122,7 @@ private static class StatisticsCollector {
   @SneakyThrows
   void testServerWithMultiplyClients() {
     LoggerManager.enable(StringSslNetworkLoadTest.class, LoggerLevel.INFO);
-    LoggerManager.enable(AbstractSslNetworkPacketReader.class, LoggerLevel.DEBUG);
+    //LoggerManager.enable(AbstractSslNetworkPacketReader.class, LoggerLevel.DEBUG);
 
     var serverConfig = SimpleServerNetworkConfig
         .builder()
@@ -136,7 +136,7 @@ void testServerWithMultiplyClients() {
     ScheduledExecutorService scheduledExecutor = Executors.newSingleThreadScheduledExecutor();
 
     int clientCount = 1;
-    int messagesPerIteration = 300;
+    int messagesPerIteration = 20;
     int expectedMessages = clientCount * messagesPerIteration * MAX_ITERATIONS;
 
     var finalWaiter = new CountDownLatch(2);
@@ -214,6 +214,8 @@ private static void initReceivedMessagesTracker(
   }
 
   private static StringWritableNetworkPacket newMessage(int minMessageLength, int maxMessageLength) {
-    return new StringWritableNetworkPacket(StringUtils.generate(minMessageLength, maxMessageLength));
+    String generate = StringUtils.generate(minMessageLength, maxMessageLength);
+    return new StringWritableNetworkPacket("a".repeat(generate.length()));
+    //return new StringWritableNetworkPacket(StringUtils.generate(minMessageLength, maxMessageLength));
   }
 }

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

@@ -1,6 +1,7 @@
 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/packet/impl/AbstractNetworkPacketWriter.java

@@ -4,6 +4,7 @@
 import static javasabr.rlib.network.util.NetworkUtils.hexDump;
 
 import java.io.IOException;
+import java.nio.BufferOverflowException;
 import java.nio.ByteBuffer;
 import java.nio.channels.AsynchronousSocketChannel;
 import java.nio.channels.CompletionHandler;
@@ -116,7 +117,6 @@ public boolean tryToSendNextPacket() {
     }
 
     serializedToChannelPacketHandler.accept(nextPacket);
-
     return startedWriting;
   }
 
@@ -143,9 +143,25 @@ protected ByteBuffer serialize(WritableNetworkPacket packet) {
       ByteBuffer second = bufferAllocator.takeBuffer(totalSize);
       firstWriteTempBuffer = first;
       secondWriteTempBuffer = second;
-      return serialize(resultPacket, expectedLength, totalSize, first, second);
+      try {
+        return serialize(resultPacket, expectedLength, totalSize, first, second);
+      } catch (BufferOverflowException ex) {
+        log.error(ex);
+        bufferAllocator.putBuffer(first);
+        bufferAllocator.putBuffer(second);
+        firstWriteTempBuffer = null;
+        secondWriteTempBuffer = null;
+        throw new RuntimeException(ex);
+      }
     } else {
-      return serialize(resultPacket, expectedLength, totalSize, firstWriteBuffer, secondWriteBuffer);
+      try {
+        return serialize(resultPacket, expectedLength, totalSize, firstWriteBuffer, secondWriteBuffer);
+      } catch (BufferOverflowException ex) {
+        log.error(ex);
+        firstWriteBuffer.clear();
+        secondWriteBuffer.clear();
+        throw new RuntimeException(ex);
+      }
     }
   }
 
@@ -359,29 +375,25 @@ protected void handleFailedWritingData(Throwable exception, WritableNetworkPacke
 
   @Override
   public void close() {
-
     bufferAllocator
         .putWriteBuffer(firstWriteBuffer)
         .putWriteBuffer(secondWriteBuffer);
-
     clearTempBuffers();
-
     writingBuffer = EMPTY_BUFFER;
   }
 
   protected void clearTempBuffers() {
 
-    var secondWriteTempBuffer = this.secondWriteTempBuffer;
     var firstWriteTempBuffer = this.firstWriteTempBuffer;
+    if (firstWriteTempBuffer != null) {
+      this.firstWriteTempBuffer = null;
+      bufferAllocator.putBuffer(firstWriteTempBuffer);
+    }
 
+    var secondWriteTempBuffer = this.secondWriteTempBuffer;
     if (secondWriteTempBuffer != null) {
       this.secondWriteTempBuffer = null;
       bufferAllocator.putBuffer(secondWriteTempBuffer);
     }
-
-    if (firstWriteTempBuffer != null) {
-      this.firstWriteTempBuffer = null;
-      bufferAllocator.putBuffer(firstWriteTempBuffer);
-    }
   }
 }

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

@@ -84,17 +84,17 @@ protected void handleReceivedData(int receivedBytes, ByteBuffer readingBuffer) {
   protected int readPackets(ByteBuffer readingBuffer) {
     ByteBuffer networkBuffer = moveDataToNetworkBuffer(readingBuffer);
     HandshakeStatus handshakeStatus = sslEngine.getHandshakeStatus();
-    if (SslUtils.isReadyToDecrypt(handshakeStatus)) {
+    if (SslUtils.isReadyToCrypt(handshakeStatus)) {
       return decryptAndRead(networkBuffer);
     } else {
       return doHandshake(networkBuffer, networkBuffer.limit());
     }
   }
 
-  private ByteBuffer moveDataToNetworkBuffer(ByteBuffer readingBuffer) {
+  protected ByteBuffer moveDataToNetworkBuffer(ByteBuffer readingBuffer) {
     ByteBuffer sslNetworkBuffer = sslNetworkBuffer();
-    int freeSpace = sslNetworkBuffer.capacity() - sslNetworkBuffer.limit();
-    if (freeSpace >= readingBuffer.limit()) {
+    int availableSpace = sslNetworkBuffer.capacity() - sslNetworkBuffer.limit();
+    if (availableSpace >= readingBuffer.limit()) {
       BufferUtils.appendAndClear(sslNetworkBuffer, readingBuffer);
     } else {
       sslNetworkBuffer = increaseNetworkBuffer(readingBuffer.limit());
@@ -103,7 +103,7 @@ private ByteBuffer moveDataToNetworkBuffer(ByteBuffer readingBuffer) {
     return sslNetworkBuffer;
   }
 
-  protected int doHandshake(ByteBuffer receivedBuffer, int receivedBytes) {
+  protected int doHandshake(ByteBuffer networkBuffer, int receivedBytes) {
     HandshakeStatus handshakeStatus = sslEngine.getHandshakeStatus();
     while (SslUtils.needToProcess(handshakeStatus)) {
       log.debug(handshakeStatus, "Do handshake with status:[%s] "::formatted);
@@ -123,13 +123,13 @@ protected int doHandshake(ByteBuffer receivedBuffer, int receivedBytes) {
             sslEngine.closeOutbound();
             handshakeStatus = sslEngine.getHandshakeStatus();
             break;
-          } else if (!receivedBuffer.hasRemaining()) {
-            receivedBuffer.clear().limit(0);
+          } else if (!networkBuffer.hasRemaining()) {
+            cleanNetworkBuffer(networkBuffer);
             return SKIP_READ_PACKETS;
           }
           try {
-            log.debug(receivedBuffer, buff -> "Try to unwrap data:\n" + hexDump(buff));
-            result = sslEngine.unwrap(receivedBuffer, EMPTY_BUFFERS);
+            log.debug(networkBuffer, buff -> "Try to unwrap data:\n" + hexDump(buff));
+            result = sslEngine.unwrap(networkBuffer, EMPTY_BUFFERS);
             handshakeStatus = result.getHandshakeStatus();
             log.debug(handshakeStatus, "Handshake status:[%s] after unwrapping"::formatted);
           } catch (SSLException sslException) {
@@ -167,21 +167,16 @@ protected int doHandshake(ByteBuffer receivedBuffer, int receivedBytes) {
           break;
         }
         case NEED_WRAP: {
-          log.debug("Send command to wrap data");
+          log.debug(remoteAddress(), "[%s] Send command to wrap data"::formatted);
           packetWriter.accept(SslWritableNetworkPacket.getInstance());
-          receivedBuffer.clear().limit(0);
+          cleanNetworkBuffer(networkBuffer);
           return SKIP_READ_PACKETS;
         }
         case NEED_TASK: {
-          Runnable task;
-          while ((task = sslEngine.getDelegatedTask()) != null) {
-            log.debug(task, "Execute SSL Engine's task:[%s]"::formatted);
-            task.run();
-          }
-          handshakeStatus = sslEngine.getHandshakeStatus();
+          handshakeStatus = SslUtils.executeSslTasks(sslEngine);
           log.debug(handshakeStatus, "Handshake status:[%s] after engine tasks"::formatted);
-          if (handshakeStatus == HandshakeStatus.NEED_UNWRAP && !receivedBuffer.hasRemaining()) {
-            receivedBuffer.clear().limit(0);
+          if (handshakeStatus == HandshakeStatus.NEED_UNWRAP && !networkBuffer.hasRemaining()) {
+            cleanNetworkBuffer(networkBuffer);
             return SKIP_READ_PACKETS;
           }
           break;
@@ -192,16 +187,16 @@ protected int doHandshake(ByteBuffer receivedBuffer, int receivedBytes) {
       }
     }
 
-    if (!receivedBuffer.hasRemaining()) {
+    if (!networkBuffer.hasRemaining()) {
       // if buffer is empty and status is FINISHED then we can notify writer
       if (handshakeStatus == HandshakeStatus.FINISHED) {
         packetWriter.accept(SslWritableNetworkPacket.getInstance());
       }
-      receivedBuffer.clear().limit(0);
+      cleanNetworkBuffer(networkBuffer);
       return SKIP_READ_PACKETS;
     }
 
-    return decryptAndRead(receivedBuffer);
+    return decryptAndRead(networkBuffer);
   }
 
   protected int decryptAndRead(ByteBuffer receivedBuffer) {
@@ -251,7 +246,7 @@ protected int decryptAndRead(ByteBuffer receivedBuffer) {
     return total;
   }
 
-  private synchronized ByteBuffer increaseNetworkBuffer(int extra) {
+  protected synchronized ByteBuffer increaseNetworkBuffer(int extra) {
     ByteBuffer current = sslNetworkBuffer();
     int newSize = (int) Math.max(current.capacity() * 1.3, current.capacity() + extra);
     sslNetworkBuffer = NetworkUtils
@@ -260,7 +255,7 @@ private synchronized ByteBuffer increaseNetworkBuffer(int extra) {
     return sslNetworkBuffer;
   }
 
-  private synchronized void increaseDataBuffer() {
+  protected synchronized void increaseDataBuffer() {
     int newSize = sslEngine
         .getSession()
         .getApplicationBufferSize();
@@ -281,4 +276,8 @@ public void close() {
         .putBuffer(sslNetworkBuffer);
     super.close();
   }
+
+  protected static void cleanNetworkBuffer(ByteBuffer networkBuffer) {
+    networkBuffer.clear().limit(0);
+  }
 }