Merge pull request #52 from JavaSaBr/Extend-API-v7

Extend API v7

Author: JavaSaBr

README.md

@@ -14,7 +14,7 @@ repositories {
 }
 
 ext {
-   rlibVersion = "10.0.alpha7"
+   rlibVersion = "10.0.alpha8"
 }
 
 dependencies {

build.gradle

@@ -1,4 +1,4 @@
-rootProject.version = "10.0.alpha7"
+rootProject.version = "10.0.alpha8"
 group = 'javasabr.rlib'
 
 allprojects {

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/loadTest/java/javasabr/rlib/network/StringNetworkLoadTest.java

@@ -41,7 +41,7 @@ private static class TestClient implements AutoCloseable {
 
     String clientId = "Client_%s".formatted(ID_FACTORY.incrementAndGet());
     NetworkConfig networkConfig = NetworkConfig.SimpleNetworkConfig.builder()
-        .groupName(clientId)
+        .threadGroupName(clientId)
         .writeBufferSize(256)
         .readBufferSize(256)
         .pendingBufferSize(512)
@@ -67,7 +67,7 @@ void connectAndSendMessages(
         ThreadUtils.sleep(random.nextInt(5000));
         StringDataConnection connection = network.connect(serverAddress);
 
-        connection.onReceive((serverConnection, packet) -> statistics
+        connection.onReceiveValidPacket((serverConnection, packet) -> statistics
             .receivedServerPackersPerSecond()
             .accumulate(1));
 
@@ -141,7 +141,7 @@ void testServerWithMultiplyClients() {
     InetSocketAddress serverAddress = serverNetwork.start();
 
     serverNetwork.onAccept(accepted -> accepted
-        .onReceive((connection, packet) -> {
+        .onReceiveValidPacket((connection, packet) -> {
           StringReadableNetworkPacket<StringDataConnection> receivedPacket = (StringReadableNetworkPacket<StringDataConnection>) packet;
           statistics
               .receivedClientPackersPerSecond()

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

@@ -44,7 +44,7 @@ private static class TestClient implements AutoCloseable {
 
     String clientId = "Client_%s".formatted(ID_FACTORY.incrementAndGet());
     NetworkConfig networkConfig = NetworkConfig.SimpleNetworkConfig.builder()
-        .groupName(clientId)
+        .threadGroupName(clientId)
         .writeBufferSize(256)
         .readBufferSize(256)
         .pendingBufferSize(512)
@@ -70,7 +70,7 @@ void connectAndSendMessages(
         ThreadUtils.sleep(random.nextInt(5000));
         StringDataSslConnection connection = network.connect(serverAddress);
 
-        connection.onReceive((serverConnection, packet) -> statistics
+        connection.onReceiveValidPacket((serverConnection, packet) -> statistics
             .receivedServerPackersPerSecond()
             .accumulate(1));
 
@@ -150,7 +150,7 @@ void testServerWithMultiplyClients() {
     InetSocketAddress serverAddress = serverNetwork.start();
 
     serverNetwork.onAccept(accepted -> accepted
-        .onReceive((connection, packet) -> {
+        .onReceiveValidPacket((connection, packet) -> {
           StringReadableNetworkPacket<StringDataSslConnection> receivedPacket = (StringReadableNetworkPacket<StringDataSslConnection>) packet;
           statistics
               .receivedClientPackersPerSecond()

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

@@ -13,10 +13,10 @@
  */
 public interface Connection<C extends Connection<C>> {
 
-  record ReceivedPacketEvent<C, R>(C connection, R packet) {
+  record ReceivedPacketEvent<C, R>(C connection, R packet, boolean valid) {
     @Override
     public String toString() {
-      return "[" + connection + "|" + packet + ']';
+      return "[" + connection + '|' + packet + '|' + valid + ']';
     }
   }
 
@@ -53,9 +53,14 @@ public String toString() {
   CompletableFuture<Boolean> sendWithFeedback(WritableNetworkPacket<C> packet);
 
   /**
-   * Register a consumer to handle received packets.
+   * Register a consumer to handle received valid packets.
    */
-  void onReceive(BiConsumer<C, ? super ReadableNetworkPacket<C>> consumer);
+  void onReceiveValidPacket(BiConsumer<C, ? super ReadableNetworkPacket<C>> consumer);
+
+  /**
+   * Register a consumer to handle received invalid packets.
+   */
+  void onReceiveInvalidPacket(BiConsumer<C, ? super ReadableNetworkPacket<C>> consumer);
 
   /**
    * Get a stream of received packet events.
@@ -72,15 +77,29 @@ default <R extends ReadableNetworkPacket<C>> Flux<ReceivedPacketEvent<C, R>> rec
   }
 
   /**
-   * Get a stream of received packets.
+   * Get a stream of received valid packets.
+   */
+  Flux<? extends ReadableNetworkPacket<C>> receivedValidPackets();
+
+  /**
+   * Get a stream of received invalid packets.
    */
-  Flux<? extends ReadableNetworkPacket<C>> receivedPackets();
+  Flux<? extends ReadableNetworkPacket<C>> receivedInvalidPackets();
+
+  /**
+   * Get a stream of received valid packets with expected type.
+   */
+  default <R extends ReadableNetworkPacket<C>> Flux<R> receivedValidPackets(Class<R> packetType) {
+    return receivedValidPackets()
+        .filter(packetType::isInstance)
+        .map(networkPacket -> (R) networkPacket);
+  }
 
   /**
-   * Get a stream of received packets with expected type.
+   * Get a stream of received invalid packets with expected type.
    */
-  default <R extends ReadableNetworkPacket<C>> Flux<R> receivedPackets(Class<R> packetType) {
-    return receivedPackets()
+  default <R extends ReadableNetworkPacket<C>> Flux<R> receivedInvalidPackets(Class<R> packetType) {
+    return receivedInvalidPackets()
         .filter(packetType::isInstance)
         .map(networkPacket -> (R) networkPacket);
   }

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) {