KeepAlive.java

import java.util.concurrent.atomic.AtomicInteger;

import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;

import io.vertx.core.Vertx;
import io.vertx.core.http.HttpClient;
import io.vertx.core.http.HttpClientOptions;
import io.vertx.core.logging.Logger;
import io.vertx.core.logging.LoggerFactory;
import io.vertx.core.net.NetServer;
import io.vertx.core.net.NetServerOptions;
import io.vertx.ext.unit.Async;
import io.vertx.ext.unit.TestContext;
import io.vertx.ext.unit.junit.VertxUnitRunner;

/**
 * 
 */

/**
 * @author <a href="http://oss.lehmann.cx/">Alexander Lehmann</a>
 *
 */
@RunWith(VertxUnitRunner.class)
public class KeepAlive {

  private static final Logger log = LoggerFactory.getLogger(KeepAlive.class);
  private final Vertx vertx = Vertx.vertx();
  private NetServer netServer;

  private AtomicInteger open = new AtomicInteger(0);
  private AtomicInteger finished = new AtomicInteger(0);

  @Test
  public void test(TestContext testContext) {
    log.info("starting test");

    HttpClientOptions clientOptions =
        new HttpClientOptions()
        .setDefaultHost("localhost")
        .setDefaultPort(8000)
        .setKeepAlive(true);

    HttpClient client = vertx.createHttpClient(clientOptions);

    for(int i=0;i<100;i++) {
      startRequest(testContext, testContext.async(), client);
      log.info("started #"+open.addAndGet(1));
    }
  }

  /**
   * @param testContext
   * @param async
   * @param client
   */
  private void startRequest(TestContext testContext, Async async, HttpClient client) {
    client.get("/", resp -> {
      log.info("request handler");
      resp.bodyHandler(buffer -> log.info("recieved body \""+buffer+"\""));
      resp.endHandler(v -> {
        log.info("finished "+finished.addAndGet(1)+" of "+open.get());
        async.complete();
      });
      resp.exceptionHandler(th -> testContext.fail(th));
    })
    .exceptionHandler(Throwable::printStackTrace)
    .end();
  }

  @Before
  public void before(TestContext testContext) {
    Async async = testContext.async();
    NetServerOptions serverOptions = new NetServerOptions();

    netServer = vertx.createNetServer(serverOptions)
        .connectHandler(ns -> {
          log.info("accepted new connection");
          vertx.setTimer(1000, t1 -> {
            ns.write("HTTP/1.1 200 OK\n" + 
                "Content-Type: text/plain\n" + 
                "Content-Length: 4\n" + 
                "Connection: keep-alive\n" + 
                "\n" + 
                "xxx\n");
            vertx.setTimer(1000, t2 -> {
              ns.write("HTTP/1.1 200 OK\n" + 
                  "Content-Type: text/plain\n" + 
                  "Content-Length: 4\n" + 
                  "Connection: close\n" + 
                  "\n" + 
                  "xxx\n");
              ns.close();
            });
          });
        })
        .listen(8000, result -> {
          if (result.succeeded()) {
            async.complete();
          } else {
            testContext.fail(result.cause());
          }
        });
    log.info("started netserver");
  }

  @After
  public void after() {
    netServer.close();
    log.info("stopped netserver");
  }

}