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(12000, 15000); // 10240
+                  StringWritableNetworkPacket message = newMessage(10, 100); // 10240
                   connection.send(message);
                 }, delay, TimeUnit.MILLISECONDS);
             tasks.add(schedule);

rlib-network/src/main/java/javasabr/rlib/network/packet/WritableNetworkPacket.java

@@ -2,6 +2,7 @@
 
 import java.nio.ByteBuffer;
 
+
 /**
  * Interface to implement a writable packet.
  *

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

@@ -34,15 +34,15 @@
 public abstract class AbstractNetworkPacketWriter<W extends WritableNetworkPacket, C extends Connection<?, W>>
     implements NetworkPacketWriter {
 
-  final CompletionHandler<Integer, WritableNetworkPacket> writeHandler = new CompletionHandler<>() {
+  final CompletionHandler<Integer, @Nullable WritableNetworkPacket> writeHandler = new CompletionHandler<>() {
 
     @Override
-    public void completed(Integer result, WritableNetworkPacket packet) {
+    public void completed(Integer result, @Nullable WritableNetworkPacket packet) {
       handleSuccessfulWritingData(result, packet);
     }
 
     @Override
-    public void failed(Throwable exc, WritableNetworkPacket packet) {
+    public void failed(Throwable exc, @Nullable WritableNetworkPacket packet) {
       handleFailedWritingData(exc, packet);
     }
   };
@@ -64,7 +64,7 @@ public void failed(Throwable exc, WritableNetworkPacket packet) {
   volatile ByteBuffer writingBuffer = EMPTY_BUFFER;
 
   final Runnable updateActivityFunction;
-  final Supplier<@Nullable WritableNetworkPacket> nextWritePacketSupplier;
+  final Supplier<@Nullable WritableNetworkPacket> writablePacketProvider;
   final ObjBoolConsumer<WritableNetworkPacket> sentPacketHandler;
   final Consumer<WritableNetworkPacket> serializedToChannelPacketHandler;
 
@@ -82,7 +82,7 @@ public AbstractNetworkPacketWriter(
     this.firstWriteBuffer = bufferAllocator.takeWriteBuffer();
     this.secondWriteBuffer = bufferAllocator.takeWriteBuffer();
     this.updateActivityFunction = updateActivityFunction;
-    this.nextWritePacketSupplier = packetProvider;
+    this.writablePacketProvider = packetProvider;
     this.serializedToChannelPacketHandler = serializedToChannelPacketHandler;
     this.sentPacketHandler = sentPacketHandler;
   }
@@ -97,29 +97,44 @@ public boolean tryToSendNextPacket() {
       return false;
     }
 
-    WritableNetworkPacket nextPacket = nextWritePacketSupplier.get();
-    if (nextPacket == null) {
-      writing.set(false);
-      return false;
+    boolean startedWriting = false;
+
+    try {
+      startedWriting = tryToSendNextPacketImpl();
+    } catch (Exception ex) {
+      log.error(ex);
     }
 
-    boolean startedWriting = false;
-    ByteBuffer resultBuffer = serialize(nextPacket);
-
-    if (resultBuffer.limit() != 0) {
-      writingBuffer = resultBuffer;
-      log.debug(remoteAddress(), resultBuffer,
-          (address, buff) -> "[%s] Write to channel data:\n" + hexDump(buff));
-      socketChannel.write(resultBuffer, nextPacket, writeHandler);
-      startedWriting = true;
-    } else {
+    if (!startedWriting) {
       writing.set(false);
     }
 
-    serializedToChannelPacketHandler.accept(nextPacket);
     return startedWriting;
   }
 
+  protected boolean tryToSendNextPacketImpl() {
+    for (var nextPacket = writablePacketProvider.get();
+         nextPacket != null; nextPacket = writablePacketProvider.get()) {
+      ByteBuffer resultBuffer = serialize(nextPacket);
+      boolean startedWriting = writeBuffer(resultBuffer, nextPacket);
+      serializedToChannelPacketHandler.accept(nextPacket);
+      if (startedWriting) {
+        return true;
+      }
+    }
+    return false;
+  }
+
+  protected boolean writeBuffer(ByteBuffer resultBuffer, @Nullable WritableNetworkPacket nextPacket) {
+    if (resultBuffer.limit() == 0) {
+      return false;
+    }
+    writingBuffer = resultBuffer;
+    log.debug(remoteAddress(), resultBuffer, (address, buff) -> "[%s] Write to channel data:\n" + hexDump(buff));
+    socketChannel.write(resultBuffer, nextPacket, writeHandler);
+    return true;
+  }
+
   /**
    * Serializes the network packet to buffer.
    *
@@ -324,11 +339,13 @@ protected ByteBuffer writeHeader(ByteBuffer buffer, int value, int headerSize) {
    * @param wroteBytes the count of wrote bytes.
    * @param packet the sent packet.
    */
-  protected void handleSuccessfulWritingData(Integer wroteBytes, WritableNetworkPacket packet) {
+  protected void handleSuccessfulWritingData(Integer wroteBytes, @Nullable WritableNetworkPacket packet) {
     updateActivityFunction.run();
 
     if (wroteBytes == -1) {
-      sentPacketHandler.accept(packet, false);
+      if (packet != null) {
+        sentPacketHandler.accept(packet, false);
+      }
       connection.close();
       return;
     }
@@ -343,13 +360,12 @@ protected void handleSuccessfulWritingData(Integer wroteBytes, WritableNetworkPa
       log.debug(remoteAddress(), wroteBytes, "[%s] Finished writing [%s] bytes"::formatted);
     }
 
-    sentPacketHandler.accept(packet, true);
+    if (packet != null) {
+      sentPacketHandler.accept(packet, true);
+    }
 
     if (writing.compareAndSet(true, false)) {
-      // if we have temp buffers, we can remove it after finishing writing a packet
-      if (firstWriteTempBuffer != null || secondWriteTempBuffer != null) {
-        clearTempBuffers();
-      }
+      clearTempBuffers();
       tryToSendNextPacket();
     }
   }
@@ -360,14 +376,15 @@ protected void handleSuccessfulWritingData(Integer wroteBytes, WritableNetworkPa
    * @param exception the exception.
    * @param packet the packet.
    */
-  protected void handleFailedWritingData(Throwable exception, WritableNetworkPacket packet) {
+  protected void handleFailedWritingData(Throwable exception, @Nullable WritableNetworkPacket packet) {
     log.error(new RuntimeException("Failed writing packet:" + packet, exception));
     if (exception instanceof IOException) {
       connection.close();
       return;
     }
     if (!connection.closed()) {
       if (writing.compareAndSet(true, false)) {
+        clearTempBuffers();
         tryToSendNextPacket();
       }
     }
@@ -383,6 +400,7 @@ public void close() {
   }
 
   protected void clearTempBuffers() {
+    this.writingBuffer = EMPTY_BUFFER;
 
     var firstWriteTempBuffer = this.firstWriteTempBuffer;
     if (firstWriteTempBuffer != null) {

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

@@ -78,6 +78,21 @@ public boolean tryToSendNextPacket() {
     return super.tryToSendNextPacket();
   }
 
+  @Override
+  protected boolean tryToSendNextPacketImpl() {
+    HandshakeStatus status = sslEngine.getHandshakeStatus();
+    if (SslUtils.isReadyToCrypt(status)) {
+      return super.tryToSendNextPacketImpl();
+    }
+
+    ByteBuffer dataToSend = doHandshake(SslWritableNetworkPacket.getInstance());
+    if (dataToSend != null) {
+      return writeBuffer(dataToSend, null);
+    }
+
+    throw new IllegalStateException("Unexpected SSL state");
+  }
+
   @Override
   protected ByteBuffer serialize(WritableNetworkPacket packet) {
     HandshakeStatus status = sslEngine.getHandshakeStatus();
@@ -158,7 +173,6 @@ protected ByteBuffer doHandshake(WritableNetworkPacket packet) {
       switch (handshakeStatus) {
         case NEED_WRAP: {
           try {
-            // check result
             result = sslEngine.wrap(EMPTY_BUFFERS, sslNetworkBuffer.clear());
             handshakeStatus = result.getHandshakeStatus();
           } catch (SSLException sslException) {
@@ -168,13 +182,14 @@ protected ByteBuffer doHandshake(WritableNetworkPacket packet) {
             handshakeStatus = sslEngine.getHandshakeStatus();
             break;
           }
+
           switch (result.getStatus()) {
             case OK:
               sslNetworkBuffer.flip();
-              if (handshakeStatus == HandshakeStatus.NEED_WRAP) {
+/*              if (handshakeStatus == HandshakeStatus.NEED_WRAP) {
                 log.debug("Send command to wrap data again");
                 queueAtFirst.accept(SslWritableNetworkPacket.getInstance());
-              }
+              }*/
               log.debug(sslNetworkBuffer, result, (buf, res) -> "Send wrapped data:\n" + hexDump(buf, res));
               return sslNetworkBuffer;
             case BUFFER_OVERFLOW: {