Reactor Netty Reference Guide Directory

The original address

Reactor Netty provides an easy-to-use, easy-to-configure TcpClient. It hides most of the Netty functionality required to create TCP clients and adds Reactive Streams back pressure.

4.1. Connect and disconnect

To connect a TCP client to a given endpoint, you must create and configure an instance of TcpClient. By default, host is localhost and POST is 12012. Here is an example of creating a TcpClient:

Github.com/reactor/rea…

import reactor.netty.Connection;
import reactor.netty.tcp.TcpClient;

public class Application {

	public static void main(String[] args) {
		Connection connection =
				TcpClient.create()      / / < 1 >
				         .connectNow(); / / < 2 >connection.onDispose() .block(); }}Copy the code

<1> The configuration operation used to create a TcpClient instance.

<2> Block the connection and wait for its initialization to complete.

The returned Connection object provides simple connection-related apis, including disposeNow(), which will be called to shut down the client in a blocking manner.

4.4.1. The Host and the Port

To connect to a specific host and port, you can configure the TCP client in the following way. The following is an example:

Github.com/reactor/rea…

import reactor.netty.Connection;
import reactor.netty.tcp.TcpClient;

public class Application {

	public static void main(String[] args) {
		Connection connection =
				TcpClient.create()
				         .host("example.com") / / < 1 >
				         .port(80)            / / < 2 >.connectNow(); connection.onDispose() .block(); }}Copy the code

<1> Configure the TCP host

<2> Configure the TCP port

4.2. Pre-initialization

By default, TcpClient initializes resources only when they are needed. This means that connect Operation takes extra time to initialize the load:

  • Event loop group

  • Host name resolver

  • Native transport library (when using Native transport)

  • Native library for security (when using OpenSsl)

When you need to preload these resources, you can configure TcpClient as follows:

Github.com/reactor/rea…

import reactor.core.publisher.Mono;
import reactor.netty.Connection;
import reactor.netty.tcp.TcpClient;

public class Application {

	public static void main(String[] args) {
		TcpClient tcpClient =
				TcpClient.create()
				         .host("example.com")
				         .port(80)
				         .handle((inbound, outbound) -> outbound.sendString(Mono.just("hello")));

		tcpClient.warmup() / / < 1 >
		         .block();

		Connection connection = tcpClient.connectNow(); / / < 2 >connection.onDispose() .block(); }}Copy the code

<1> Initialize and load the event loop group, host name resolver, Native transport library and Native library for security

<2> Host name resolution is performed when connecting to the remote node

4.3. Write out the data

If you want to send data to an existing endpoint, you must add an I/O handler. The I/O processor can write out data via NettyOutbound.

Github.com/reactor/rea…

import reactor.core.publisher.Mono;
import reactor.netty.Connection;
import reactor.netty.tcp.TcpClient;

public class Application {

	public static void main(String[] args) {
		Connection connection =
				TcpClient.create()
				         .host("example.com")
				         .port(80)
				         .handle((inbound, outbound) -> outbound.sendString(Mono.just("hello"))) / / < 1 >.connectNow(); connection.onDispose() .block(); }}Copy the code

<1> sends a Hello string to this endpoint.

4.4. Consumption data

If you want to receive data from an existing endpoint, you must add an I/O handler. The I/O processor can read data via NettyInbound. The following is an example:

Github.com/reactor/rea…

import reactor.netty.Connection;
import reactor.netty.tcp.TcpClient;

public class Application {

	public static void main(String[] args) {
		Connection connection =
				TcpClient.create()
				         .host("example.com")
				         .port(80)
				         .handle((inbound, outbound) -> inbound.receive().then()) / / < 1 >.connectNow(); connection.onDispose() .block(); }}Copy the code

<1> Receives data sent from an existing endpoint

4.5. Lifecycle callbacks

The following lifecycle callback parameters are provided to you to extend TcpClient:

Callback Description
doAfterResolve Called after successful resolution of the remote address.
doOnChannelInit Called when a channel is initialized.
doOnConnect Called when a channel is about to connect.
doOnConnected Called when a channel is already connected.
doOnDisconnected Called when a channel breaks.
doOnResolve Called when the remote address is about to be resolved.
doOnResolveError Called in case remote address resolution fails.

Here is an example of using the doOnConnected and doOnChannelInit callback:

Github.com/reactor/rea…

import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.ReadTimeoutHandler;
import reactor.netty.Connection;
import reactor.netty.tcp.TcpClient;
import java.util.concurrent.TimeUnit;

public class Application {

	public static void main(String[] args) {
		Connection connection =
				TcpClient.create()
				         .host("example.com")
				         .port(80)
				         .doOnConnected(conn ->
				             conn.addHandler(new ReadTimeoutHandler(10, TimeUnit.SECONDS))) / / < 1 >
				         .doOnChannelInit((observer, channel, remoteAddress) ->
				             channel.pipeline()
				                    .addFirst(new LoggingHandler("reactor.netty.examples")))/ / < 2 >.connectNow(); connection.onDispose() .block(); }}Copy the code

<1> Added a ReadTimeoutHandler to Netty Pipeline when a channel is connected.

<2> Added a LoggingHandler to Netty Pipeline when initializing a channel.

4.6. Configure the TCP layer

This section describes three ways to configure the TCP layer:

  • Channel Options

  • Wire Logger

  • Event Loop Group

4.6.1. Channel Options

By default, the TCP client is configured with the following options:

. /.. /.. /reactor-netty-core/src/main/java/reactor/netty/tcp/TcpClientConnect.java

TcpClientConnect(ConnectionProvider provider) {
    this.config = new TcpClientConfig(
        provider,
        Collections.singletonMap(ChannelOption.AUTO_READ, false),
        () -> AddressUtils.createUnresolved(NetUtil.LOCALHOST.getHostAddress(), DEFAULT_PORT));
}
Copy the code

To add a new option or modify an existing option, you can use the following methods:

Github.com/reactor/rea…

import io.netty.channel.ChannelOption;
import reactor.netty.Connection;
import reactor.netty.tcp.TcpClient;

public class Application {

	public static void main(String[] args) {
		Connection connection =
				TcpClient.create()
				         .host("example.com")
				         .port(80)
				         .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 10000) .connectNow(); connection.onDispose() .block(); }}Copy the code

You can find out more about Nettychannel Options by following the link below:

  • ChannelOption

  • Socket Options

4.6.2. Wire Logger

Reactor Netty provides wire logging for checking point-to-point traffic. By default, line logging is turned off. If you want to open it, you must put the log reactor.net ty. TCP. The TcpClient Settings for the DEBUG level and configuration as follows:

Github.com/reactor/rea…

import reactor.netty.Connection;
import reactor.netty.tcp.TcpClient;

public class Application {

	public static void main(String[] args) {
		Connection connection =
				TcpClient.create()
				         .wiretap(true) / / < 1 >
				         .host("example.com")
				         .port(80) .connectNow(); connection.onDispose() .block(); }}Copy the code

<1> Enable line recording

By default, line records output content using AdvancedBytebufType #HEX_DUMP. You can also change it to AdvancedBytebufType #SIMPLE or AdvancedBytebufType #TEXTUAL by configuring TcpClient:

Github.com/reactor/rea…

import io.netty.handler.logging.LogLevel;
import reactor.netty.Connection;
import reactor.netty.tcp.TcpClient;
import reactor.netty.transport.logging.AdvancedByteBufFormat;

public class Application {

	public static void main(String[] args) {
		Connection connection =
				TcpClient.create()
				         .wiretap("logger-name", LogLevel.DEBUG, AdvancedByteBufFormat.TEXTUAL) / / < 1 >
				         .host("example.com")
				         .port(80) .connectNow(); connection.onDispose() .block(); }}Copy the code

<1> Open line recording and use AdvancedBytebufForm #TEXTUAL to export the content.

4.6.3. The Event Loop Group

By default, TCP clients use an “Event Loop Group” where the number of worker threads is equal to the number of processors available at initialization (but a minimum of 4). You can also modify the configuration using one of the methods LoopResource#create.

The default Event Loop Group configuration is as follows:

. /.. /.. /reactor-netty-core/src/main/java/reactor/netty/ReactorNetty.java

/** * Default worker thread count, fallback to available processor * (but with a minimum value of 4) */
public static final String IO_WORKER_COUNT = "reactor.netty.ioWorkerCount";
/** * Default selector thread count, fallback to -1 (no selector thread) */
public static final String IO_SELECT_COUNT = "reactor.netty.ioSelectCount";
/** * Default worker thread count for UDP, fallback to available processor * (but with a minimum value of 4) */
public static final String UDP_IO_THREAD_COUNT = "reactor.netty.udp.ioThreadCount";
/** * Default quiet period that guarantees that the disposal of the underlying LoopResources * will not happen, fallback to 2 seconds. */
public static final String SHUTDOWN_QUIET_PERIOD = "reactor.netty.ioShutdownQuietPeriod";
/** * Default maximum amount of time to wait until the disposal of the underlying LoopResources * regardless if a task was submitted during the quiet period, fallback to 15 seconds. */
public static final String SHUTDOWN_TIMEOUT = "reactor.netty.ioShutdownTimeout";

/** * Default value whether the native transport (epoll, kqueue) will be preferred, * fallback it will be preferred when available */
public static final String NATIVE = "reactor.netty.native";
Copy the code

If you need to modify these Settings, you can also configure them as follows:

Github.com/reactor/rea…

import reactor.netty.Connection;
import reactor.netty.resources.LoopResources;
import reactor.netty.tcp.TcpClient;

public class Application {

	public static void main(String[] args) {
		LoopResources loop = LoopResources.create("event-loop".1.4.true);

		Connection connection =
				TcpClient.create()
				         .host("example.com")
				         .port(80) .runOn(loop) .connectNow(); connection.onDispose() .block(); }}Copy the code

4.7. The connection pool

By default, TCP clients use a “fixed” connection pool, with a maximum number of channels of 500 and a maximum number of fetch registration requests in the queue of 1000 (see system properties below for the rest of the configuration). This means that if someone tries to fetch a channel from the pool, but no channel is available in the pool, a new channel will be created. When the number of channels in the pool reaches its maximum, new channel acquisition operations are delayed until an available channel is returned to the pool again.

. /.. /.. /reactor-netty-core/src/main/java/reactor/netty/ReactorNetty.java

/** * Default max connections. Fallback to * available number of processors (but with a minimum value of 16) */
public static final String POOL_MAX_CONNECTIONS = "reactor.netty.pool.maxConnections";
/** * Default acquisition timeout (milliseconds) before error. If -1 will never wait to * acquire before opening a new *  connection in an unbounded fashion. Fallback 45 seconds */
public static final String POOL_ACQUIRE_TIMEOUT = "reactor.netty.pool.acquireTimeout";
/** * Default max idle time, fallback - max idle time is not specified. */
public static final String POOL_MAX_IDLE_TIME = "reactor.netty.pool.maxIdleTime";
/** * Default max life time, fallback - max life time is not specified. */
public static final String POOL_MAX_LIFE_TIME = "reactor.netty.pool.maxLifeTime";
/** * Default leasing strategy (fifo, lifo), fallback to fifo. * 
       
    
         * 
        
  • fifo - The connection selection is first in, first out
    •  * 
        
  • lifo - The connection selection is last in, first out
    •  * 
       
 */
public static final String POOL_LEASING_STRATEGY = "reactor.netty.pool.leasingStrategy";
/**
 * Default {@code getPermitsSamplingRate} (between 0d and 1d (percentage))
 * to be used with a {@link SamplingAllocationStrategy}.
 * This strategy wraps a {@link PoolBuilder#sizeBetween(int, int) sizeBetween} {@link AllocationStrategy}
 * and samples calls to {@link AllocationStrategy#getPermits(int)}.
 * Fallback - sampling is not enabled.
 */
public static final String POOL_GET_PERMITS_SAMPLING_RATE = "reactor.netty.pool.getPermitsSamplingRate";
/**
 * Default {@code returnPermitsSamplingRate} (between 0d and 1d (percentage))
 * to be used with a {@link SamplingAllocationStrategy}.
 * This strategy wraps a {@link PoolBuilder#sizeBetween(int, int) sizeBetween} {@link AllocationStrategy}
 * and samples calls to {@link AllocationStrategy#returnPermits(int)}.
 * Fallback - sampling is not enabled.
 */
public static final String POOL_RETURN_PERMITS_SAMPLING_RATE = "reactor.netty.pool.returnPermitsSamplingRate";
Copy the code

If you need to disable connection pooling, you can use the following configuration:

Github.com/reactor/rea…

import reactor.netty.Connection;
import reactor.netty.tcp.TcpClient;

public class Application {

	public static void main(String[] args) {
		Connection connection =
				TcpClient.newConnection()
				         .host("example.com")
				         .port(80) .connectNow(); connection.onDispose() .block(); }}Copy the code

If you need to set a specific idle time for a channel in the connection pool, you can use the following configuration:

Github.com/reactor/rea…

import reactor.netty.Connection;
import reactor.netty.resources.ConnectionProvider;
import reactor.netty.tcp.TcpClient;
import java.time.Duration;

public class Application {

	public static void main(String[] args) {
		ConnectionProvider provider =
				ConnectionProvider.builder("fixed")
				                  .maxConnections(50)
				                  .pendingAcquireTimeout(Duration.ofMillis(30000))
				                  .maxIdleTime(Duration.ofMillis(60))
				                  .build();

		Connection connection =
				TcpClient.create(provider)
				         .host("example.com")
				         .port(80) .connectNow(); connection.onDispose() .block(); }}Copy the code

It is prudent to use a connection pool with a high maximum number of connections when you expect a high load. Because you can encounter reactor.net ty. HTTP. Client. PrematureCloseException abnormalities, the fundamental reason is that too many simultaneous connections to the opened/acquired as a result of operation is the Timeout “Connect”.

4.7.1. Measure

The pooled ConnectionProvider provides built-in integration with Micrometer. It exposes all prefixes to reactor.net ty. Connection. The measurement of the provider.

Pooled ConnectionProvider metrics

Measure names type describe
reactor.netty.connection.provider.total.connections Gauge The number of all connections, both active and idle
reactor.netty.connection.provider.active.connections Gauge The number of connections that have been successfully acquired and are in use
reactor.netty.connection.provider.idle.connections Gauge Number of idle connections
reactor.netty.connection.provider.pending.connections Gauge Number of requests waiting for available connections

Here is an example of enabling integration metrics:

Github.com/reactor/rea…

import reactor.netty.Connection;
import reactor.netty.resources.ConnectionProvider;
import reactor.netty.tcp.TcpClient;

public class Application {

	public static void main(String[] args) {
		ConnectionProvider provider =
				ConnectionProvider.builder("fixed")
				                  .maxConnections(50)
				                  .metrics(true) / / < 1 >
				                  .build();

		Connection connection =
				TcpClient.create(provider)
				         .host("example.com")
				         .port(80) .connectNow(); connection.onDispose() .block(); }}Copy the code

<1> Start the built-in integrated Micrometer

4.8. The SSL and TLS

When you need to use SSL or TLS, you can use the configuration listed below. By default, sslprovider.openssl is used if OpenSSL is available. Otherwise, use sslprovider.jdk. – Dio.net ty. Can be set by SslContextBuilder or handler. SSL. NoOpenSsl = true to switch.

Here is an example using SslContextBuilder:

Github.com/reactor/rea…

import io.netty.handler.ssl.SslContextBuilder;
import reactor.netty.Connection;
import reactor.netty.tcp.TcpClient;

public class Application {

	public static void main(String[] args) {
		SslContextBuilder sslContextBuilder = SslContextBuilder.forClient();

		Connection connection =
				TcpClient.create()
				         .host("example.com")
				         .port(443) .secure(spec -> spec.sslContext(sslContextBuilder)) .connectNow(); connection.onDispose() .block(); }}Copy the code

4.8.1. Server name identification

By default, TCP clients send the remote host name as the SNI server name. To modify the default Settings, you can configure the TCP client as follows:

Github.com/reactor/rea…

import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslContextBuilder;
import reactor.netty.Connection;
import reactor.netty.tcp.TcpClient;

import javax.net.ssl.SNIHostName;

public class Application {

	public static void main(String[] args) throws Exception {
		SslContext sslContext = SslContextBuilder.forClient().build();

		Connection connection =
				TcpClient.create()
				         .host("127.0.0.1")
				         .port(8080)
				         .secure(spec -> spec.sslContext(sslContext)
				                             .serverNames(new SNIHostName("test.com"))) .connectNow(); connection.onDispose() .block(); }}Copy the code

4.9. Proxy support

TCP clients support proxy functionality provided by Netty and provide a specific “non-proxy host” approach through the ProxyProvider builder. Here is an example of using ProxyProvider:

Github.com/reactor/rea…

import reactor.netty.Connection;
import reactor.netty.transport.ProxyProvider;
import reactor.netty.tcp.TcpClient;

public class Application {

	public static void main(String[] args) {
		Connection connection =
				TcpClient.create()
				         .host("example.com")
				         .port(80)
				         .proxy(spec -> spec.type(ProxyProvider.Proxy.SOCKS4)
				                            .host("proxy")
				                            .port(8080)
				                            .nonProxyHosts("localhost")) .connectNow(); connection.onDispose() .block(); }}Copy the code

4.10. Measurement

The TCP client supports built-in integration with Micrometer. It exposes all metrics prefixed with Reactor.net ty.tcp.client.

The following table provides information about TCP client metrics:

Measure names type describe
reactor.netty.tcp.client.data.received DistributionSummary The amount of data received, in bytes
reactor.netty.tcp.client.data.sent DistributionSummary The amount of data sent, in bytes
reactor.netty.tcp.client.errors Counter The number of errors that occur
reactor.netty.tcp.client.tls.handshake.time Timer Time spent in TLS handshake
reactor.netty.tcp.client.connect.time Timer The time taken to connect to the remote address
reactor.netty.tcp.client.address.resolver Timer The time taken to resolve the remote address

The following additional metrics are also available:

Pooled ConnectionProvider metrics

Measure names type describe
reactor.netty.connection.provider.total.connections Gauge The number of all connections, both active and idle
reactor.netty.connection.provider.active.connections Gauge The number of connections that have been successfully acquired and are in use
reactor.netty.connection.provider.idle.connections Gauge Number of idle connections
reactor.netty.connection.provider.pending.connections Gauge Number of requests waiting for available connections

ByteBufAllocator measure

Measure names type describe
reactor.netty.bytebuf.allocator.used.heap.memory Gauge The number of bytes of heap memory
reactor.netty.bytebuf.allocator.used.direct.memory Gauge The number of bytes of out-of-heap memory
reactor.netty.bytebuf.allocator.used.heap.arenas Gauge The number of heap memory (when usedPooledByteBufAllocatorWhen)
reactor.netty.bytebuf.allocator.used.direct.arenas Gauge The number of out-of-heap memory (when usedPooledByteBufAllocatorWhen)
reactor.netty.bytebuf.allocator.used.threadlocal.caches Gauge The number of threadLocal caches (when usedPooledByteBufAllocatorWhen)
reactor.netty.bytebuf.allocator.used.tiny.cache.size Gauge Tiny cache size (when usedPooledByteBufAllocatorWhen)
reactor.netty.bytebuf.allocator.used.small.cache.size Gauge Small cache size (when usedPooledByteBufAllocatorWhen)
reactor.netty.bytebuf.allocator.used.normal.cache.size Gauge General cache size (when usedPooledByteBufAllocatorWhen)
reactor.netty.bytebuf.allocator.used.chunk.size Gauge Block size of an area (when usedPooledByteBufAllocatorWhen)

Here is an example of enabling integration metrics:

Github.com/reactor/rea…

import reactor.netty.Connection;
import reactor.netty.tcp.TcpClient;

public class Application {

	public static void main(String[] args) {
		Connection connection =
				TcpClient.create()
				         .host("example.com")
				         .port(80)
				         .metrics(true) / / < 1 >.connectNow(); connection.onDispose() .block(); }}Copy the code

<1> Start the built-in integrated Micrometer

If you want TCP client metrics to integrate with systems other than Micrometer or you want to provide your own integration with Micrometer to add your own metrics logger, you can do this as follows:

Github.com/reactor/rea…

import reactor.netty.Connection;
import reactor.netty.channel.ChannelMetricsRecorder;
import reactor.netty.tcp.TcpClient;

import java.net.SocketAddress;
import java.time.Duration;

public class Application {

	public static void main(String[] args) {
		Connection connection =
				TcpClient.create()
				         .host("example.com")
				         .port(80)
				         .metrics(true, CustomChannelMetricsRecorder::new) / / < 1 >.connectNow(); connection.onDispose() .block(); }}Copy the code

<1> Enable TCP client metrics and provide ChannelMetricsRecorder implementation.

4.11.Unix domain sockets

TCP clients support Unix domain sockets (UDS) when using local transports.

The following is an example of using UDS:

Github.com/reactor/rea…

import io.netty.channel.unix.DomainSocketAddress;
import reactor.netty.Connection;
import reactor.netty.tcp.TcpClient;

public class Application {

	public static void main(String[] args) {
		Connection connection =
				TcpClient.create()
				         .remoteAddress(() -> new DomainSocketAddress("/tmp/test.sock")) / / < 1 >.connectNow(); connection.onDispose() .block(); }}Copy the code

<1> Specifies the DomainSocketAddress to use

4.12. Host name resolution

By default, TcpClient uses Netty’s domain name query mechanism to resolve domain names asynchronously. Used to replace the JVM’s built-in blocking parser.

To modify the default Settings, you can configure the TcpClient as follows:

Github.com/reactor/rea…

import reactor.netty.Connection;
import reactor.netty.tcp.TcpClient;

import java.time.Duration;

public class Application {

	public static void main(String[] args) {
		Connection connection =
				TcpClient.create()
				         .host("example.com")
				         .port(80)
				         .resolver(spec -> spec.queryTimeout(Duration.ofMillis(500))) / / < 1 >.connectNow(); connection.onDispose() .block(); }}Copy the code

<1> The timeout period for each DNS query executed by the parser is set to 500ms.

The following list shows the configurations available:

The name of the configuration describe
cacheMaxTimeToLive Maximum lifetime of DNS resource record cache (unit: second). If the survival time of DNS resource records returned by the DNS server is greater than the maximum survival time. The parser will ignore the lifetime from the DNS server and use this maximum lifetime. The default isInteger.MAX_VALUE
cacheMinTimeToLive Minimum lifetime of DNS resource record cache (unit: second). If the DNS server returns a DNS resource record with a live time less than this minimum, the parser ignores the live time from the DNS server and uses the minimum. Default: 0.
cacheNegativeTimeToLive Cache time of DNS query failure (unit: second). Default: 0.
disableOptionalRecord Disable the optional automatic inclusion setting that tries to tell the remote DNS server parser how much data can be read per response. By default, this setting is enabled.
disableRecursionDesired Used to specify whether the parser queries DNS with the expected recursive query (RD) flag. By default, this setting is enabled.
maxPayloadSize Sets the size (in bytes) of the datagram packet buffer. Default: 4096
maxQueriesPerResolve Set the maximum number of DNS queries that can be sent for resolving host names. Default: 16
ndots Sets the number of dots that must appear in the name when an absolute query is initialized. Default value: -1 (used to determine the value of the Unix operating system; otherwise, 1 is used).
queryTimeout Set the timeout (in milliseconds) for each DNS query by the parser. Default: 5000
resolvedAddressTypes Protocol family list for resolving addresses.
roundRobinSelection To enable theDnsNameResolvertheAddressResolverGroupTo support random selection of destination addresses when multiple addresses are provided by the naming server. seeRoundRobinDnsAddressResolverGroup. The default:DnsAddressResolverGroup.
runOn In a givenLoopResourcesTo perform communication with the DNS server. By default, LoopResources is only used on the client.
searchDomains The list of search fields for the parser. By default, the list of valid search domains is the system DNS search domain used.
trace The logger and log level used by the parser to generate detailed trace information in the event of a parsing failure.

Sometimes, you might want to switch to a parser built into the JVM. You can configure the TcpClient as follows:

Github.com/reactor/rea…

import io.netty.resolver.DefaultAddressResolverGroup;
import reactor.netty.Connection;
import reactor.netty.tcp.TcpClient;

public class Application {

	public static void main(String[] args) {
		Connection connection =
				TcpClient.create()
				         .host("example.com")
				         .port(80)
				         .resolver(DefaultAddressResolverGroup.INSTANCE) / / < 1 >.connectNow(); connection.onDispose() .block(); }}Copy the code

<1> set to a parser built into the JVM.

Suggest Edit to “TCP Client”

Reactor Netty Reference Guide Directory

Copyright notice: If you need to reprint, please bring the link to this article, note the source and this statement. Otherwise, legal liability will be investigated. https://www.immuthex.com/posts/reactor-netty-reference-guide/tcp-client