improving network module

Author: JavaSaBr

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

@@ -122,7 +122,7 @@ void testServerWithMultiplyClients() {
 
  var serverConfig = SimpleServerNetworkConfig
  .builder()
- .threadGroupSize(10)
+ .threadGroupMaxSize(10)
  .writeBufferSize(1024)
  .readBufferSize(1024)
  .pendingBufferSize(2048)

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

@@ -129,7 +129,7 @@ void testServerWithMultiplyClients() {
 
  var serverConfig = SimpleServerNetworkConfig
  .builder()
- .threadGroupSize(10)
+ .threadGroupMaxSize(10)
  .writeBufferSize(1024)
  .readBufferSize(1024)
  .pendingBufferSize(2048)

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

@@ -1,12 +1,18 @@
 package javasabr.rlib.network;
 
+import java.util.concurrent.ScheduledExecutorService;
+
 /**
  * The interface to implement an asynchronous network.
  *
  * @author JavaSaBr
  */
 public interface Network<C extends Connection<C>> {
 
+ ScheduledExecutorService scheduledExecutor();
+
+ NetworkConfig config();
+
  /**
  * Shutdown this network.
  */

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

@@ -28,6 +28,8 @@ class SimpleNetworkConfig implements NetworkConfig {
  private int pendingBufferSize = 4096;
  @Builder.Default
  private int writeBufferSize = 2048;
+ @Builder.Default
+ private int retryDelayInMs = 1000;
  }
 
  NetworkConfig DEFAULT_CLIENT = new NetworkConfig() {
@@ -46,34 +48,46 @@ default String threadGroupName() {
  }
 
  /**
- * Get size of buffer with used to collect received data from network.
+ * Get a group name of scheduling network threads.
+ */
+ default String scheduledThreadGroupName() {
+ return "ScheduledNetworkThread";
+ }
+
+ /**
+ * Get size of buffer with will be used to collect received data from network.
  */
  default int readBufferSize() {
  return 2048;
  }
 
  /**
- * Get size of buffer with pending data. Pending buffer allows to construct a packet with bigger data than
+ * Gets size of buffer with pending reading data. Pending buffer allows to construct a packet with bigger data part than
  * {@link #readBufferSize()}. It should be at least 2x of {@link #readBufferSize()}
- *
- * @return the pending buffer's size.
  */
  default int pendingBufferSize() {
  return readBufferSize() * 2;
  }
 
  /**
- * Get size of buffer which used to serialize packets to bytes.
+ * Gets a size of buffer which will be used for packet serialization.
  */
  default int writeBufferSize() {
  return 2048;
  }
 
+ /**
+ * Gets a timeout for retry read/write operation.
+ */
+ default int retryDelayInMs() {
+ return 1000;
+ }
+
  default ByteOrder byteOrder() {
  return ByteOrder.BIG_ENDIAN;
  }
 
- default boolean isDirectByteBuffer() {
+ default boolean useDirectByteBuffer() {
  return false;
  }
 }

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

@@ -32,21 +32,27 @@ class SimpleServerNetworkConfig implements ServerNetworkConfig {
  @Builder.Default
  private int writeBufferSize = 2048;
  @Builder.Default
- private int threadGroupSize = 1;
+ private int retryDelayInMs = 1000;
+ @Builder.Default
+ private int threadGroupMinSize = 1;
+ @Builder.Default
+ private int threadGroupMaxSize = 1;
+ @Builder.Default
+ private int scheduledThreadGroupSize = 1;
  @Builder.Default
  private int threadPriority = Thread.NORM_PRIORITY;
  }
 
  ServerNetworkConfig DEFAULT_SERVER = new ServerNetworkConfig() {
 
  @Override
- public int threadGroupMinSize() {
- return 1;
+ public String threadGroupName() {
+ return "ServerNetworkThread";
  }
 
  @Override
- public String threadGroupName() {
- return "ServerNetworkThread";
+ public String scheduledThreadGroupName() {
+ return "ServerScheduledNetworkThread";
  }
  };
 
@@ -64,6 +70,13 @@ default int threadGroupMaxSize() {
  return threadGroupMinSize();
  }
 
+ /**
+ * Get a size of network scheduled thread executor.
+ */
+ default int scheduledThreadGroupSize() {
+ return 1;
+ }
+
  /**
  * Get a thread constructor which should be used to create network threads.
  */

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

@@ -2,5 +2,7 @@
 
 public interface UnsafeConnection<C extends UnsafeConnection<C>> extends Connection<C> {
 
+ Network<?> network();
+
  void onConnected();
 }

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

@@ -6,9 +6,12 @@
 import java.nio.channels.CompletionHandler;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.BiFunction;
 import javasabr.rlib.common.util.AsyncUtils;
+import javasabr.rlib.common.util.GroupThreadFactory;
 import javasabr.rlib.common.util.ThreadUtils;
 import javasabr.rlib.common.util.Utils;
 import javasabr.rlib.network.Connection;
@@ -38,6 +41,9 @@ public class DefaultClientNetwork<C extends Connection<C>> extends AbstractNetwo
 
  final AtomicBoolean connecting;
 
+ @Getter
+ final ScheduledExecutorService scheduledExecutor;
+
  @Nullable
  @Getter(AccessLevel.PROTECTED)
  volatile CompletableFuture<C> pendingConnection;
@@ -51,6 +57,8 @@ public DefaultClientNetwork(
  BiFunction<Network<C>, AsynchronousSocketChannel, C> channelToConnection) {
  super(config, channelToConnection);
  this.connecting = new AtomicBoolean(false);
+ this.scheduledExecutor = Executors
+ .newSingleThreadScheduledExecutor(new GroupThreadFactory(config.scheduledThreadGroupName()));
  log.info(config, DefaultClientNetwork::buildConfigDescription);
  }
 
@@ -85,7 +93,7 @@ public CompletableFuture<C> connectAsync(InetSocketAddress serverAddress) {
  @Override
  public void completed(@Nullable Void result, DefaultClientNetwork<C> network) {
  SocketAddress remoteAddress = NetworkUtils.getRemoteAddress(channel);
- log.info(remoteAddress, "Connected to server:[%s]"::formatted);
+ log.info(remoteAddress, "[%s] Connected to server."::formatted);
  asyncResult.complete(channelToConnection.apply(network, channel));
  }
 

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

@@ -51,8 +51,9 @@ public WritablePacketWithFeedback(CompletableFuture<Boolean> attachment, Writabl
 
  @Getter
  final String remoteAddress;
-
+ @Getter
  final Network<C> network;
+
  final BufferAllocator bufferAllocator;
  final AsynchronousSocketChannel channel;
  final Deque<WritableNetworkPacket<C>> pendingPackets;

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

@@ -7,7 +7,9 @@
 import javasabr.rlib.network.NetworkConfig;
 import lombok.AccessLevel;
 import lombok.CustomLog;
+import lombok.Getter;
 import lombok.RequiredArgsConstructor;
+import lombok.experimental.Accessors;
 import lombok.experimental.FieldDefaults;
 
 /**
@@ -17,8 +19,11 @@
  */
 @CustomLog
 @RequiredArgsConstructor
+@Accessors(fluent = true, chain = false)
 @FieldDefaults(level = AccessLevel.PROTECTED, makeFinal = true)
 public abstract class AbstractNetwork<C extends Connection<C>> implements Network<C> {
+
+ @Getter
  NetworkConfig config;
  BiFunction<Network<C>, AsynchronousSocketChannel, C> channelToConnection;
 }

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

@@ -28,7 +28,7 @@ public DefaultBufferAllocator(NetworkConfig config) {
  public ByteBuffer takeReadBuffer() {
  int bufferSize = config.readBufferSize();
  log.debug(bufferSize, "Allocate new read buffer with size:[%s]"::formatted);
- return config.isDirectByteBuffer()
+ return config.useDirectByteBuffer()
  ? ByteBuffer.allocateDirect(bufferSize)
  : ByteBuffer
  .allocate(bufferSize)
@@ -40,7 +40,7 @@ public ByteBuffer takeReadBuffer() {
  public ByteBuffer takePendingBuffer() {
  int bufferSize = config.pendingBufferSize();
  log.debug(bufferSize, "Allocate new pending buffer with size:[%s]"::formatted);
- return config.isDirectByteBuffer()
+ return config.useDirectByteBuffer()
  ? ByteBuffer.allocateDirect(bufferSize)
  : ByteBuffer
  .allocate(bufferSize)
@@ -52,7 +52,7 @@ public ByteBuffer takePendingBuffer() {
  public ByteBuffer takeWriteBuffer() {
  int bufferSize = config.writeBufferSize();
  log.debug(bufferSize, "Allocate new write buffer with size:[%s]"::formatted);
- return config.isDirectByteBuffer()
+ return config.useDirectByteBuffer()
  ? ByteBuffer.allocateDirect(bufferSize)
  : ByteBuffer
  .allocate(bufferSize)
@@ -63,7 +63,7 @@ public ByteBuffer takeWriteBuffer() {
  @Override
  public ByteBuffer takeBuffer(int bufferSize) {
  log.debug(bufferSize, "Allocate new buffer with size:[%s]"::formatted);
- return config.isDirectByteBuffer()
+ return config.useDirectByteBuffer()
  ? ByteBuffer.allocateDirect(bufferSize)
  : ByteBuffer
  .allocate(bufferSize)