Ensure only one TCP connection using Netty4Http

Question

I am trying to send a large number of HTTPS requests using Netty4Http component. Here is a sample code to test this:

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import org.apache.camel.CamelContext;
import org.apache.camel.CamelExecutionException;
import org.apache.camel.ProducerTemplate;
import org.apache.camel.component.netty4.http.NettyHttpComponent;
import org.apache.camel.impl.DefaultCamelContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class DataWriter {

    private static final Logger LOG = LoggerFactory.getLogger(DataWriter.class);
    private static DataWriter dataWriter;
    private final CamelContext context;
    private final Map<String, Object> headers = new HashMap<>();
 // private NettyHttpEndpoint nettyHttpEndpoint;
    private ProducerTemplate template;
    private String endpoint;

public static void createDataWriterInstance() {
    if (dataWriter == null) {
        dataWriter = new DataWriter();
    }
}

public static DataWriter getDataWriterInstance() {
    return dataWriter;
}

private DataWriter() {
    this.context = new DefaultCamelContext();
    try {
        final NettyHttpComponent nettyHttpComponent = this.context.getComponent("netty4-http",
                org.apache.camel.component.netty4.http.NettyHttpComponent.class);
        this.context.addComponent("fcpnettyhttpComponent", nettyHttpComponent);
        this.template = this.context.createProducerTemplate();
        this.headers.put("Content-Type", "application/json");
        this.headers.put("Connection", "keep-alive");
        this.headers.put("CamelHttpMethod", "POST");
        String trustCertificate = "&ssl=true&passphrase=" + "123456" + "&keyStoreFile="
                + "C:/Users/jpisaac/certs/publicKey.store"
                + "&trustStoreFile=C:/Users/jpisaac/certs/publicKey.store" ;

        this.endpoint = "netty4-http:"+ "https://xx.xx.xx.xx:8443/server"
                + "?useByteBuf=true&disableStreamCache=true&connectTimeout=30000&requestTimeout=30000&reuseChannel=true"
                + "&keepAlive=true&tcpNoDelay=true&sync=false&reuseAddress=true&sendBufferSize=1000"
                + trustCertificate;
        this.template.start();
        this.context.start();
    } catch (final Exception e) {
        LOG.error("Exception while starting Camel context ", e);
    }
}

public void sendData(final String message) {
    try {
        CompletableFuture<Object> future=this.template.asyncRequestBodyAndHeaders(this.endpoint, message, this.headers);
        System.err.println("Sent data "+message);
    } catch (final CamelExecutionException e) {
        LOG.error("Error while sending data", e);
    }
}

public void stop() {
    try {
        this.template.stop();
        this.context.stop();
    } catch (final Exception e) {
        LOG.error("Exception while stopping Camel context ", e);
    }
}

public static void main(String[] args) throws Exception {
    createDataWriterInstance();
    DataWriter test = getDataWriterInstance();
    int i = 0;
    while (i<50) {
        test.sendData("Hello " + i++);
    }
    while (true) {}
}
}

This code works, however we see that multiple ports are opened while sending the data in async mode. This can be verified by checking on Wireshark.

Also when we analyze the program on JVisualVM we can see that several NettyClientTCPWorker and ProducerTemplate threads are created. I see that we can control the number of worker threads by the workerCount setting.

I have a restraint on the number of ports that I open on my client machine to send data to the server. I will need to keep it at a configurable value(usually 1).

How can I ensure that only one port is opened on the client machine and still use the async mode?

I tried setting the producerPoolMaxActive property to 1. Now only one port gets opened, but that also means that only one request is sent. Looks like a port is opened for each request that is sent. This is something I need to avoid.

[Update] I have added Connection: keep-alive in the headers, but that did not help. I think the core issue is that a new connection is being opened for each request. I see this in the logs:

2017-07-03 11:25:32.367 [Camel (camel-1) thread #0 - ProducerTemplate] DEBUG o.a.c.component.netty4.NettyProducer.openConnection(NettyProducer.java:436) - Created new TCP client bootstrap connecting to 10.194.242.10:8443 with options: Bootstrap(BootstrapConfig(group: NioEventLoopGroup, channelFactory: NioSocketChannel.class, options: {SO_KEEPALIVE=true, TCP_NODELAY=true, SO_REUSEADDR=true, CONNECT_TIMEOUT_MILLIS=30000}, handler: org.apache.camel.component.netty4.http.HttpClientInitializerFactory@4d814f1e, resolver: io.netty.resolver.DefaultAddressResolverGroup@488dff6f))
2017-07-03 11:25:32.366 [Camel (camel-1) thread #4 - ProducerTemplate] DEBUG o.a.c.component.netty4.NettyProducer.openConnection(NettyProducer.java:436) - Created new TCP client bootstrap connecting to 10.194.242.10:8443 with options: Bootstrap(BootstrapConfig(group: NioEventLoopGroup, channelFactory: NioSocketChannel.class, options: {SO_KEEPALIVE=true, TCP_NODELAY=true, SO_REUSEADDR=true, CONNECT_TIMEOUT_MILLIS=30000}, handler: org.apache.camel.component.netty4.http.HttpClientInitializerFactory@4d814f1e, resolver: io.netty.resolver.DefaultAddressResolverGroup@488dff6f))
2017-07-03 11:25:32.366 [Camel (camel-1) thread #3 - ProducerTemplate] DEBUG o.a.c.component.netty4.NettyProducer.openConnection(NettyProducer.java:436) - Created new TCP client bootstrap connecting to 10.194.242.10:8443 with options: Bootstrap(BootstrapConfig(group: NioEventLoopGroup, channelFactory: NioSocketChannel.class, options: {SO_KEEPALIVE=true, TCP_NODELAY=true, SO_REUSEADDR=true, CONNECT_TIMEOUT_MILLIS=30000}, handler: org.apache.camel.component.netty4.http.HttpClientInitializerFactory@4d814f1e, resolver: io.netty.resolver.DefaultAddressResolverGroup@488dff6f))
2017-07-03 11:25:32.367 [Camel (camel-1) thread #1 - ProducerTemplate] DEBUG o.a.c.component.netty4.NettyProducer.openConnection(NettyProducer.java:436) - Created new TCP client bootstrap connecting to 10.194.242.10:8443 with options: Bootstrap(BootstrapConfig(group: NioEventLoopGroup, channelFactory: NioSocketChannel.class, options: {SO_KEEPALIVE=true, TCP_NODELAY=true, SO_REUSEADDR=true, CONNECT_TIMEOUT_MILLIS=30000}, handler: org.apache.camel.component.netty4.http.HttpClientInitializerFactory@4d814f1e, resolver: io.netty.resolver.DefaultAddressResolverGroup@488dff6f))
2017-07-03 11:25:32.366 [Camel (camel-1) thread #2 - ProducerTemplate] DEBUG o.a.c.component.netty4.NettyProducer.openConnection(NettyProducer.java:436) - Created new TCP client bootstrap connecting to 10.194.242.10:8443 with options: Bootstrap(BootstrapConfig(group: NioEventLoopGroup, channelFactory: NioSocketChannel.class, options: {SO_KEEPALIVE=true, TCP_NODELAY=true, SO_REUSEADDR=true, CONNECT_TIMEOUT_MILLIS=30000}, handler: org.apache.camel.component.netty4.http.HttpClientInitializerFactory@4d814f1e, resolver: io.netty.resolver.DefaultAddressResolverGroup@488dff6f))

Answer 1

Add HTTP header Connection: keep-alive, otherwise the server should close the connection after each request.