finish client ssl network

Author: JavaSaBr

rlib-network/src/main/java/com/ss/rlib/network/NetworkFactory.java

@@ -118,6 +118,25 @@ public final class NetworkFactory {
         );
     }
 
+    /**
+     * Create string packet based asynchronous secure client network.
+     *
+     * @param networkConfig   the network config.
+     * @param bufferAllocator the buffer allocator.
+     * @param sslContext      the ssl context.
+     * @return the client network.
+     */
+    public static @NotNull ClientNetwork<StringDataSSLConnection> newStringDataSSLClientNetwork(
+        @NotNull NetworkConfig networkConfig,
+        @NotNull BufferAllocator bufferAllocator,
+        @NotNull SSLContext sslContext
+    ) {
+        return newClientNetwork(
+            networkConfig,
+            (network, channel) -> new StringDataSSLConnection(network, channel, bufferAllocator, sslContext, true)
+        );
+    }
+
     /**
      * Create string packet based asynchronous server network.
      *

rlib-network/src/main/java/com/ss/rlib/network/client/impl/DefaultClientNetwork.java

@@ -5,6 +5,8 @@
 import com.ss.rlib.common.concurrent.util.ThreadUtils;
 import com.ss.rlib.common.util.AsyncUtils;
 import com.ss.rlib.common.util.Utils;
+import com.ss.rlib.logger.api.Logger;
+import com.ss.rlib.logger.api.LoggerManager;
 import com.ss.rlib.network.Connection;
 import com.ss.rlib.network.Network;
 import com.ss.rlib.network.NetworkConfig;
@@ -31,7 +33,9 @@
  */
 public class DefaultClientNetwork<C extends Connection<?, ?>> extends AbstractNetwork<C> implements ClientNetwork<C> {
 
-    protected final AtomicBoolean connecting;
+    protected static final Logger LOGGER = LoggerManager.getLogger(DefaultClientNetwork.class);
+
+    protected final @NotNull AtomicBoolean connecting;
 
     protected volatile @Nullable CompletableFuture<C> pendingConnection;
     protected volatile @Getter @Nullable C currentConnection;
@@ -42,6 +46,13 @@ public DefaultClientNetwork(
     ) {
         super(config, channelToConnection);
         this.connecting = new AtomicBoolean(false);
+
+        LOGGER.info(config, conf -> "Client network configuration: {\n" +
+            "  groupName: \"" + conf.getThreadGroupName() + "\",\n" +
+            "  readBufferSize: " + conf.getReadBufferSize() + ",\n" +
+            "  pendingBufferSize: " + conf.getPendingBufferSize() + ",\n" +
+            "  writeBufferSize: " + conf.getWriteBufferSize() + "\n" +
+            "}");
     }
 
     @Override

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

@@ -206,6 +206,15 @@ protected void sendImpl(@NotNull WritablePacket packet) {
         getPacketWriter().writeNextPacket();
     }
 
+    protected void queueAtFirst(@NotNull WritablePacket packet) {
+        long stamp = lock.writeLock();
+        try {
+            pendingPackets.addFirst(packet);
+        } finally {
+            lock.unlockWrite(stamp);
+        }
+    }
+
     @Override
     public @NotNull CompletableFuture<Boolean> sendWithFeedback(@NotNull W packet) {
 

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

@@ -69,6 +69,7 @@ public DefaultDataSSLConnection(
             this::onSentPacket,
             sslEngine,
             this::sendImpl,
+            this::queueAtFirst,
             packetLengthHeaderSize
         );
     }

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

@@ -92,6 +92,7 @@ protected int doHandshake(@NotNull ByteBuffer receivedBuffer, int receivedBytes)
         var handshakeStatus = sslEngine.getHandshakeStatus();
 
         while (handshakeStatus != HandshakeStatus.FINISHED && handshakeStatus != HandshakeStatus.NOT_HANDSHAKING) {
+            LOGGER.debug(handshakeStatus, status -> "Do handshake with status: " + status);
 
             SSLEngineResult result;
 
@@ -124,6 +125,7 @@ protected int doHandshake(@NotNull ByteBuffer receivedBuffer, int receivedBytes)
                         LOGGER.debug(receivedBuffer, buff -> "Try to unwrap data:\n" + hexDump(buff));
                         result = sslEngine.unwrap(receivedBuffer, EMPTY_BUFFERS);
                         handshakeStatus = result.getHandshakeStatus();
+                        LOGGER.debug(handshakeStatus, status -> "Handshake status: " + status + " after unwrapping");
                     } catch (SSLException sslException) {
                         LOGGER.error("A problem was encountered while processing the data that caused the " +
                             "SSLEngine to abort. Will try to properly close connection...");
@@ -170,20 +172,29 @@ protected int doHandshake(@NotNull ByteBuffer receivedBuffer, int receivedBytes)
 
                     handshakeStatus = sslEngine.getHandshakeStatus();
 
+                    LOGGER.debug(handshakeStatus, status -> "Handshake status: " + status + " after engine tasks");
+
                     if (handshakeStatus == HandshakeStatus.NEED_UNWRAP && !receivedBuffer.hasRemaining()) {
                         sslNetworkBuffer.clear();
                         return SKIP_READ_PACKETS;
                     }
 
-                    break;
-                case FINISHED:
-                case NOT_HANDSHAKING:
                     break;
                 default:
                     throw new IllegalStateException("Invalid SSL status: " + handshakeStatus);
             }
         }
 
+        if (!receivedBuffer.hasRemaining()) {
+
+            // if buffer is empty and status is FINISHED then we can notify writer
+            if (handshakeStatus == HandshakeStatus.FINISHED) {
+                packetWriter.accept(SSLWritablePacket.getInstance());
+            }
+
+            return SKIP_READ_PACKETS;
+        }
+
         return decryptAndRead(receivedBuffer);
     }
 

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

@@ -6,6 +6,7 @@
 import com.ss.rlib.common.function.NotNullConsumer;
 import com.ss.rlib.common.function.NullableSupplier;
 import com.ss.rlib.logger.api.Logger;
+import com.ss.rlib.logger.api.LoggerLevel;
 import com.ss.rlib.logger.api.LoggerManager;
 import com.ss.rlib.network.BufferAllocator;
 import com.ss.rlib.network.Connection;
@@ -33,6 +34,7 @@ public abstract class AbstractSSLPacketWriter<W extends WritablePacket, C extend
 
     protected final @NotNull SSLEngine sslEngine;
     protected final @NotNull NotNullConsumer<WritablePacket> packetWriter;
+    protected final @NotNull NotNullConsumer<WritablePacket> queueAtFirst;
 
     protected volatile @NotNull ByteBuffer sslNetworkBuffer;
 
@@ -45,7 +47,8 @@ public AbstractSSLPacketWriter(
         @NotNull NotNullConsumer<WritablePacket> writtenPacketHandler,
         @NotNull NotNullBiConsumer<WritablePacket, Boolean> sentPacketHandler,
         @NotNull SSLEngine sslEngine,
-        @NotNull NotNullConsumer<WritablePacket> packetWriter
+        @NotNull NotNullConsumer<WritablePacket> packetWriter,
+        @NotNull NotNullConsumer<WritablePacket> queueAtFirst
     ) {
         super(
             connection,
@@ -58,16 +61,34 @@ public AbstractSSLPacketWriter(
         );
         this.sslEngine = sslEngine;
         this.packetWriter = packetWriter;
+        this.queueAtFirst = queueAtFirst;
         this.sslNetworkBuffer = bufferAllocator.takeBuffer(sslEngine.getSession().getPacketBufferSize());
     }
 
+    @Override
+    public void writeNextPacket() {
+
+        var status = sslEngine.getHandshakeStatus();
+
+        switch (status) {
+            case NEED_UNWRAP:
+                return;
+        }
+
+        super.writeNextPacket();
+    }
+
     @Override
     protected @NotNull ByteBuffer serialize(@NotNull WritablePacket packet) {
 
         var status = sslEngine.getHandshakeStatus();
 
         if (status == HandshakeStatus.FINISHED || status == HandshakeStatus.NOT_HANDSHAKING) {
 
+            if (packet instanceof SSLWritablePacket) {
+                return EMPTY_BUFFER;
+            }
+
             SSLEngineResult result;
             try {
                 result = sslEngine.wrap(super.serialize(packet), sslNetworkBuffer.clear());
@@ -106,7 +127,8 @@ public AbstractSSLPacketWriter(
     protected @Nullable ByteBuffer doHandshake(@NotNull WritablePacket packet) throws SSLException {
 
         if (!(packet instanceof SSLWritablePacket)) {
-            throw new IllegalStateException();
+            LOGGER.debug(packet, pck -> "Return packet " + pck + " to queue as first");
+            queueAtFirst.accept(packet);
         }
 
         var handshakeStatus = sslEngine.getHandshakeStatus();
@@ -136,10 +158,11 @@ public AbstractSSLPacketWriter(
 
                             if (handshakeStatus == HandshakeStatus.NEED_WRAP) {
                                 LOGGER.debug("Send command to wrap data again");
-                                packetWriter.accept(SSLWritablePacket.getInstance());
+                                queueAtFirst.accept(SSLWritablePacket.getInstance());
                             }
 
-                            LOGGER.debug(sslNetworkBuffer,
+                            LOGGER.debug(
+                                sslNetworkBuffer,
                                 result,
                                 (buf, res) -> "Send wrapped data:\n" + hexDump(buf, res)
                             );
@@ -170,18 +193,14 @@ public AbstractSSLPacketWriter(
                     }
                     handshakeStatus = sslEngine.getHandshakeStatus();
                     break;
-                case FINISHED:
-                    break;
-                case NOT_HANDSHAKING:
-                    break;
                 case NEED_UNWRAP:
-                    return EMPTY_BUFFER;
+                    break;
                 default:
                     throw new IllegalStateException("Invalid SSL status: " + handshakeStatus);
             }
         }
 
-        return null;
+        return EMPTY_BUFFER;
     }
 
     private void increaseNetworkBuffer() {

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

@@ -30,6 +30,7 @@ public DefaultSSLPacketWriter(
         @NotNull NotNullBiConsumer<WritablePacket, Boolean> sentPacketHandler,
         @NotNull SSLEngine sslEngine,
         @NotNull NotNullConsumer<WritablePacket> packetWriter,
+        @NotNull NotNullConsumer<WritablePacket> queueAtFirst,
         int packetLengthHeaderSize
     ) {
         super(
@@ -41,7 +42,8 @@ public DefaultSSLPacketWriter(
             writtenPacketHandler,
             sentPacketHandler,
             sslEngine,
-            packetWriter
+            packetWriter,
+            queueAtFirst
         );
         this.packetLengthHeaderSize = packetLengthHeaderSize;
     }

rlib-network/src/main/java/com/ss/rlib/network/server/impl/DefaultServerNetwork.java

@@ -6,6 +6,8 @@
 import com.ss.rlib.common.util.Utils;
 import com.ss.rlib.common.util.array.Array;
 import com.ss.rlib.common.util.array.ArrayFactory;
+import com.ss.rlib.logger.api.Logger;
+import com.ss.rlib.logger.api.LoggerManager;
 import com.ss.rlib.network.Network;
 import com.ss.rlib.network.ServerNetworkConfig;
 import com.ss.rlib.network.UnsafeConnection;
@@ -34,6 +36,8 @@
 public final class DefaultServerNetwork<C extends UnsafeConnection<?, ?>> extends AbstractNetwork<C> implements
     ServerNetwork<C> {
 
+    protected static final Logger LOGGER = LoggerManager.getLogger(DefaultServerNetwork.class);
+
     private interface ServerCompletionHandler<C extends UnsafeConnection<?, ?>> extends
         CompletionHandler<AsynchronousSocketChannel, DefaultServerNetwork<C>> {}