extend API of network module, improve handling invalid received packets

Author: JavaSaBr

rlib-collections/src/test/java/javasabr/rlib/collections/array/IntArrayTest.java

@@ -2,7 +2,6 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 
-import java.util.List;
 import java.util.stream.Stream;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;

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

@@ -13,6 +13,8 @@ public interface Network<C extends Connection<C>> {
 
   NetworkConfig config();
 
+  void inNetworkThread(Runnable task);
+
   /**
    * Shutdown this network.
    */

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

@@ -1,6 +1,7 @@
 package javasabr.rlib.network;
 
 import java.nio.ByteOrder;
+import javasabr.rlib.common.util.GroupThreadFactory.ThreadConstructor;
 import lombok.Builder;
 import lombok.Getter;
 import lombok.experimental.Accessors;
@@ -18,9 +19,13 @@ public interface NetworkConfig {
   class SimpleNetworkConfig implements NetworkConfig {
 
     @Builder.Default
-    private String groupName = "NetworkThread";
+    private String threadGroupName = "NetworkThread";
     @Builder.Default
     private ByteOrder byteOrder = ByteOrder.BIG_ENDIAN;
+    @Builder.Default
+    private ThreadConstructor threadConstructor = Thread::new;
+    @Builder.Default
+    private int threadPriority = Thread.NORM_PRIORITY;
 
     @Builder.Default
     private int readBufferSize = 2048;
@@ -46,6 +51,22 @@ public String threadGroupName() {
     }
   };
 
+  /**
+   * Get a thread constructor which should be used to create network threads.
+   */
+  default ThreadConstructor threadConstructor() {
+    return Thread::new;
+  }
+
+  /**
+   * Get a priority of network threads.
+   *
+   * @return the priority of network threads.
+   */
+  default int threadPriority() {
+    return Thread.NORM_PRIORITY;
+  }
+
   /**
    * Get a group name of network threads.
    */

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

@@ -82,21 +82,4 @@ default int threadGroupMaxSize() {
   default int scheduledThreadGroupSize() {
     return 1;
   }
-
-
-  /**
-   * Get a thread constructor which should be used to create network threads.
-   */
-  default ThreadConstructor threadConstructor() {
-    return Thread::new;
-  }
-
-  /**
-   * Get a priority of network threads.
-   *
-   * @return the priority of network threads.
-   */
-  default int threadPriority() {
-    return Thread.NORM_PRIORITY;
-  }
 }

rlib-network/src/main/java/javasabr/rlib/network/client/impl/DefaultClientNetwork.java

@@ -2,10 +2,12 @@
 
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
+import java.nio.channels.AsynchronousChannelGroup;
 import java.nio.channels.AsynchronousSocketChannel;
 import java.nio.channels.CompletionHandler;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -43,6 +45,8 @@ public class DefaultClientNetwork<C extends Connection<C>> extends AbstractNetwo
 
   @Getter
   final ScheduledExecutorService scheduledExecutor;
+  final ExecutorService networkExecutor;
+  final AsynchronousChannelGroup channelGroup;
 
   @Nullable
   @Getter(AccessLevel.PROTECTED)
@@ -57,11 +61,17 @@ public DefaultClientNetwork(
       BiFunction<Network<C>, AsynchronousSocketChannel, C> channelToConnection) {
     super(config, channelToConnection);
     this.connecting = new AtomicBoolean(false);
-    this.scheduledExecutor = Executors
-        .newSingleThreadScheduledExecutor(new GroupThreadFactory(config.scheduledThreadGroupName()));
+    this.scheduledExecutor = buildScheduledExecutor(config);
+    this.networkExecutor = buildExecutor(config);
+    this.channelGroup = Utils.uncheckedGet(networkExecutor, AsynchronousChannelGroup::withThreadPool);
     log.info(config, DefaultClientNetwork::buildConfigDescription);
   }
 
+  @Override
+  public void inNetworkThread(Runnable task) {
+    networkExecutor.execute(task);
+  }
+
   @Override
   public C connect(InetSocketAddress serverAddress) {
     return connectAsync(serverAddress).join();
@@ -88,7 +98,7 @@ public CompletableFuture<C> connectAsync(InetSocketAddress serverAddress) {
     var asyncResult = new CompletableFuture<C>();
 
     @SuppressWarnings("resource")
-    var channel = Utils.uncheckedGet(AsynchronousSocketChannel::open);
+    var channel = Utils.uncheckedGet(channelGroup, AsynchronousSocketChannel::open);
     channel.connect(serverAddress, this, new CompletionHandler<>() {
       @Override
       public void completed(@Nullable Void result, DefaultClientNetwork<C> network) {
@@ -136,6 +146,33 @@ public void shutdown() {
     if (connection != null) {
       Utils.unchecked(connection, C::close);
     }
+    channelGroup.shutdown();
+    scheduledExecutor.shutdown();
+    networkExecutor.shutdown();
+  }
+
+  protected ExecutorService buildExecutor(NetworkConfig config) {
+    var threadFactory = new GroupThreadFactory(
+        config.threadGroupName(),
+        config.threadConstructor(),
+        config.threadPriority(),
+        false);
+    ExecutorService executorService = Executors.newSingleThreadScheduledExecutor(threadFactory);
+    // activate the executor
+    executorService.submit(() -> {});
+    return executorService;
+  }
+
+  protected ScheduledExecutorService buildScheduledExecutor(NetworkConfig config) {
+    var threadFactory = new GroupThreadFactory(
+        config.scheduledThreadGroupName(),
+        config.threadConstructor(),
+        config.threadPriority(),
+        false);
+    ScheduledExecutorService scheduledExecutor = Executors.newSingleThreadScheduledExecutor(threadFactory);
+    // activate the executor
+    scheduledExecutor.submit(() -> {});
+    return scheduledExecutor;
   }
 
   private static String buildConfigDescription(NetworkConfig conf) {

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

@@ -97,13 +97,13 @@ public void onConnected() {}
   @Override
   public void onReceiveValidPacket(BiConsumer<C, ? super ReadableNetworkPacket<C>> consumer) {
     validPacketSubscribers.add(consumer);
-    packetReader().startRead();
+    network.inNetworkThread(() -> packetReader().startRead());
   }
 
   @Override
   public void onReceiveInvalidPacket(BiConsumer<C, ? super ReadableNetworkPacket<C>> consumer) {
     invalidPacketSubscribers.add(consumer);
-    packetReader().startRead();
+    network.inNetworkThread(() -> packetReader().startRead());
   }
 
   @Override
@@ -129,26 +129,32 @@ protected void registerFluxOnReceivedEvents(
         packet, true));
     BiConsumer<C, ReadableNetworkPacket<C>> invalidListener =
         (connection, packet) -> sink.next(new ReceivedPacketEvent<>(connection,
-            packet, true));
+            packet, false));
 
 
-    onReceiveValidPacket(validListener);
-    onReceiveInvalidPacket(invalidListener);
+    validPacketSubscribers.add(validListener);
+    invalidPacketSubscribers.add(invalidListener);
 
-    sink.onDispose(() -> validPacketSubscribers.remove(validListener));
-    sink.onDispose(() -> invalidPacketSubscribers.remove(invalidListener));
+    sink.onDispose(() -> {
+      validPacketSubscribers.remove(validListener);
+      validPacketSubscribers.remove(invalidListener);
+    });
+
+    network.inNetworkThread(() -> packetReader().startRead());
   }
 
   protected void registerFluxOnReceivedValidPackets(FluxSink<? super ReadableNetworkPacket<C>> sink) {
     BiConsumer<C, ReadableNetworkPacket<C>> listener = (connection, packet) -> sink.next(packet);
-    onReceiveValidPacket(listener);
+    validPacketSubscribers.add(listener);
     sink.onDispose(() -> validPacketSubscribers.remove(listener));
+    network.inNetworkThread(() -> packetReader().startRead());
   }
 
   protected void registerFluxOnReceivedInvalidPackets(FluxSink<? super ReadableNetworkPacket<C>> sink) {
     BiConsumer<C, ReadableNetworkPacket<C>> listener = (connection, packet) -> sink.next(packet);
-    onReceiveInvalidPacket(listener);
+    invalidPacketSubscribers.add(listener);
     sink.onDispose(() -> invalidPacketSubscribers.remove(listener));
+    network.inNetworkThread(() -> packetReader().startRead());
   }
 
   @Nullable
@@ -222,18 +228,15 @@ public final void send(WritableNetworkPacket<C> packet) {
   }
 
   protected void sendImpl(WritableNetworkPacket<C> packet) {
-
     if (closed()) {
       return;
     }
-
     long stamp = lock.writeLock();
     try {
       pendingPackets.addLast(packet);
     } finally {
       lock.unlockWrite(stamp);
     }
-
     packetWriter().tryToSendNextPacket();
   }
 
@@ -248,15 +251,11 @@ protected void queueAtFirst(WritableNetworkPacket<C> packet) {
 
   @Override
   public CompletableFuture<Boolean> sendWithFeedback(WritableNetworkPacket<C> packet) {
-
     var asyncResult = new CompletableFuture<Boolean>();
-
     sendImpl(new WritablePacketWithFeedback<>(asyncResult, packet));
-
     if (closed()) {
       return CompletableFuture.completedFuture(Boolean.FALSE);
     }
-
     return asyncResult;
   }
 
@@ -273,11 +272,9 @@ protected void clearWaitPackets() {
   }
 
   protected void doClearWaitPackets() {
-
     for (var pendingPacket : pendingPackets) {
       handleSentPacket(pendingPacket, false);
     }
-
     pendingPackets.clear();
   }
 

rlib-network/src/main/java/javasabr/rlib/network/server/impl/DefaultServerNetwork.java

@@ -57,7 +57,7 @@ private interface ServerCompletionHandler<C extends UnsafeConnection<C>> extends
     public void completed(AsynchronousSocketChannel channel, DefaultServerNetwork<C> network) {
       var connection = network.channelToConnection.apply(network, channel);
       log.debug(connection.remoteAddress(), "Accepted new connection:[%s]"::formatted);
-      network.onAccept(connection);
+      network.consumeAccepted(connection);
       network.acceptNext();
     }
 
@@ -77,18 +77,20 @@ public void failed(Throwable exc, DefaultServerNetwork<C> network) {
 
   @Getter
   ScheduledExecutorService scheduledExecutor;
+  ExecutorService networkExecutor;
 
-  AsynchronousChannelGroup group;
+  AsynchronousChannelGroup channelGroup;
   AsynchronousServerSocketChannel channel;
   MutableArray<Consumer<? super C>> subscribers;
 
   public DefaultServerNetwork(
       ServerNetworkConfig config,
       BiFunction<Network<C>, AsynchronousSocketChannel, C> channelToConnection) {
     super(config, channelToConnection);
-    this.group = Utils.uncheckedGet(buildExecutor(config), AsynchronousChannelGroup::withThreadPool);
+    this.networkExecutor = buildExecutor(config);
+    this.channelGroup = Utils.uncheckedGet(networkExecutor, AsynchronousChannelGroup::withThreadPool);
     this.scheduledExecutor = buildScheduledExecutor(config);
-    this.channel = Utils.uncheckedGet(group, AsynchronousServerSocketChannel::open);
+    this.channel = Utils.uncheckedGet(channelGroup, AsynchronousServerSocketChannel::open);
     this.subscribers = ArrayFactory.copyOnModifyArray(Consumer.class);
     log.info(config, DefaultServerNetwork::buildConfigDescription);
   }
@@ -109,18 +111,23 @@ public InetSocketAddress start() {
     log.info(address, "Started server socket on address:[%s]"::formatted);
 
     if (!subscribers.isEmpty()) {
-      acceptNext();
+      inNetworkThread(this::acceptNext);
     }
 
     return address;
   }
 
+  @Override
+  public void inNetworkThread(Runnable task) {
+    networkExecutor.execute(task);
+  }
+
   @Override
   public <S extends ServerNetwork<C>> S start(InetSocketAddress serverAddress) {
     Utils.unchecked(channel, serverAddress, AsynchronousServerSocketChannel::bind);
     log.info(serverAddress, addr -> "Started server socket on address: " + addr);
     if (!subscribers.isEmpty()) {
-      acceptNext();
+      inNetworkThread(this::acceptNext);
     }
     return ClassUtils.unsafeNNCast(this);
   }
@@ -135,7 +142,7 @@ protected void acceptNext() {
     }
   }
 
-  protected void onAccept(C connection) {
+  protected void consumeAccepted(C connection) {
     connection.onConnected();
     subscribers
         .iterations()
@@ -145,7 +152,7 @@ protected void onAccept(C connection) {
   @Override
   public void onAccept(Consumer<? super C> consumer) {
     subscribers.add(consumer);
-    acceptNext();
+    inNetworkThread(this::acceptNext);
   }
 
   @Override
@@ -162,7 +169,9 @@ protected void registerFluxOnAccepted(FluxSink<C> sink) {
   @Override
   public void shutdown() {
     Utils.unchecked(channel, AsynchronousChannel::close);
-    group.shutdown();
+    channelGroup.shutdown();
+    scheduledExecutor.shutdown();
+    networkExecutor.shutdown();
   }
 
   protected ExecutorService buildExecutor(ServerNetworkConfig config) {