HTTP Client requests done right

Question

As advised by many, I am using a client pool - specifically the Apache PoolingHttpClientConnectionManager.

For simplicity I wrap it in my own simple singleton class. Sorry about the rather OTT Stop mechanism:

public final class HttpClientPool {
  private static final Logger log = LoggerFactory.getLogger(HttpClientPool.class);

  // Single-element enum to implement Singleton.
  private static enum Singleton {
    // Just one of me so constructor will be called once.
    Client;
    // The thread-safe client.
    private final CloseableHttpClient threadSafeClient;
    // The pool monitor.
    private final IdleConnectionMonitorThread monitor;

    // The constructor creates it - thus late
    private Singleton() {
      PoolingHttpClientConnectionManager cm = new PoolingHttpClientConnectionManager();
      // Increase max total connection to 200
      cm.setMaxTotal(200);
      // Increase default max connection per route to 20
      cm.setDefaultMaxPerRoute(20);
      // Build the client.
      threadSafeClient = HttpClients.custom()
              .setConnectionManager(cm)
              .build();
      // Start up an eviction thread.
      monitor = new IdleConnectionMonitorThread(cm);
      // Don't stop quitting.
      monitor.setDaemon(true);
      monitor.start();
    }

    public CloseableHttpClient get() {
      return threadSafeClient;
    }

  }

  public static CloseableHttpClient getClient() {
    // The thread safe client is held by the singleton.
    return Singleton.Client.get();
  }

  // Watches for stale connections and evicts them.
  private static class IdleConnectionMonitorThread extends Thread {
    // The manager to watch.
    private final PoolingHttpClientConnectionManager cm;
    // Use a BlockingQueue to stop everything.
    private final BlockingQueue<Stop> stopSignal = new ArrayBlockingQueue<Stop>(1);

    // Pushed up the queue.
    private static class Stop {
      // The return queue.
      private final BlockingQueue<Stop> stop = new ArrayBlockingQueue<Stop>(1);

      // Called by the process that is being told to stop.
      public void stopped() {
        // Push me back up the queue to indicate we are now stopped.
        stop.add(this);
      }

      // Called by the process requesting the stop.
      public void waitForStopped() throws InterruptedException {
        // Wait until the callee acknowledges that it has stopped.
        stop.take();
      }

    }

    IdleConnectionMonitorThread(PoolingHttpClientConnectionManager cm) {
      super();
      this.cm = cm;
    }

    @Override
    public void run() {
      try {
        // Holds the stop request that stopped the process.
        Stop stopRequest;
        // Every 5 seconds.
        while ((stopRequest = stopSignal.poll(5, TimeUnit.SECONDS)) == null) {
          // Close expired connections
          cm.closeExpiredConnections();
          // Optionally, close connections that have been idle too long.
          cm.closeIdleConnections(60, TimeUnit.SECONDS);
          // Look at pool stats.
          log.trace("Stats: {}", cm.getTotalStats());
        }
        // Acknowledge the stop request.
        stopRequest.stopped();
      } catch (InterruptedException ex) {
        // terminate
      }
    }

    public void shutdown() throws InterruptedException {
      log.trace("Shutting down client pool");
      // Signal the stop to the thread.
      Stop stop = new Stop();
      stopSignal.add(stop);
      // Wait for the stop to complete.
      stop.waitForStopped();
      // Close the pool - Added
      threadSafeClient.close();
      // Close the connection manager.
      cm.close();
      log.trace("Client pool shut down");
    }

  }

  public static void shutdown() throws InterruptedException {
    // Shutdown the monitor.
    Singleton.Client.monitor.shutdown();
  }

}

I use it exclusively with JSON requests:

  // General query of the website. Takes an object of type Q and returns one of class R.
  public static <Q extends JSONObject, R> R query(String urlBase, String op, Q q, Class<R> r) throws IOException {
    // The request.
    final HttpRequestBase request;
    //postRequest.addHeader("Accept-Encoding", "gzip,deflate");
    if (q != null) {
      // Prepare the post.
      HttpPost postRequest = new HttpPost(urlBase + op);
      // Get it all into a JSON string.
      StringEntity input = new StringEntity(asJSONString(q));
      input.setContentType("application/json");
      postRequest.setEntity(input);
      // Use that one.
      request = postRequest;
    } else {
      // Just get.
      request = new HttpGet(urlBase + op);
    }
    log.debug("> " + urlBase + op + (q == null ? "" : " " + q));
    // Post it and wait.
    return readResponse(request, HttpClientPool.getClient().execute(request), r);
  }
  public static <R> R readResponse(HttpRequestBase request, CloseableHttpResponse response, Class<R> r) throws IOException {
    // What was read.
    R red = null;
    try {
      // What happened?
      if (response.getStatusLine().getStatusCode() == 200) {
        // Roll out the results
        HttpEntity entity = response.getEntity();
        if (entity != null) {
          InputStream content = entity.getContent();
          try {
            // Roll it directly from the response stream.
            JsonParser rsp = getFactory().createJsonParser(content);
            // Bring back the response.
            red = rsp.readValueAs(r);
          } finally {
            // Always close the content.
            content.close();
          }
        }
      } else {
        // The finally below will clean up.
        throw new IOException("HTTP Response: " + response.getStatusLine().getStatusCode());
      }
    } finally {
      // Always close the response.
      response.close();
    }

    if (red == null) {
      log.debug("< {null}");
    } else {
      log.debug("< {}", red.getClass().isArray() ? Arrays.toString((Object[]) red) : red.toString());
    }
    return red;
  }

This seems to work fine under normal load - however, a recent high-load period caused everything to fall apart. We even interfered with other hosted apps. Not sure what caused the interference but I want to be sure this side of it is done right.

The exceptions I saw were:

org.apache.http.NoHttpResponseException: The target server failed to respond
java.net.BindException: Address already in use: connect
java.net.SocketException: No buffer space available (maximum connections reached?): connect -- Thousands of these per hour!!!

So - my questions:

1. Am I using the pool correctly?

2. Am I closing/not closing at the right time?

3. Should I reset the request (request.reset())?

4. Have I missed something?

---

Added

Spot the deliberate mistake - not closing the threadSafeClient at shutdown time. Not relevant to the issue but important. Fixed!

Also - failing to close the stream and checking the entity against null. Fixed!

Answer 1

I'm not an expert in the library or multithreading, so maybe some advises will not be applicable.

TL;DR
I didn't find anything in your code really wrong. In fact, the more I look at it, the more I can just find only nitpicking things, and I'm really forcing myself. I would find myself quite happy to maintain this code. Everything is clean, the exception managing is excellent and it's quite easy to read.

readResponse Looks clean to me. What is important when working with the request is to always "consume" the entity and/or close the request. At first, I thought that you had a problem when the request didn't return a 200. When the return code is not 200 you're not consuming the Entity which can normally block the connection, but it seems that you don't need to consume the Entity if you close() the request. In my code, I'm still using EntityUtils.consume(), but it's probably overkill.

query seems a bit weird to me. I always see either get or post not both. I find weird that query can do both, since normally it's very two different things. Despite that fact, it still really well handled.

This line is a bit hard to fully grasp at first read IMO :

return readResponse(request, HttpClientPool.getClient().execute(request), r);

I find it hard to see that this it's actually where you're making the request. It could be in it's own line, to better extract it's important role. I find it easier to debug too.

final HttpRequestBase request;
//postRequest.addHeader("Accept-Encoding", "gzip,deflate");

I guess the comment is a dead one. It could be remove since it serve no purpose. In general, I find there is a bit too much comments for my own taste, but this is basically because your code is very clean and comments are just expressing what I understand by just reading your code. (I'm really impress by the quality)

// Post it and wait.

This one is not quite true, because you could be executing a Get request and not a Post. I'm starting to think that those methods were at first working with Post request and evolve into both type of request. (base on the dead comment and this one)

 if (response.getStatusLine().getStatusCode() == 200)

Instead of 200 I would personaly use HttpStatus.OK.

Answer 2

One small correction here:

  // Close the pool - Added
  threadSafeClient.close();

Change this to:

// Close the pool - Added
Singleton.Client.threadSafeClient.close();

Explanation:

There is no threadSafeClient field in IdleConnectionMonitorThread class, thus the code fails to compile. Since this field is present in Singleton, the correct way of closing this is:

Singleton.Client.threadSafeClient.close();

This fixes the compilation error.

Answer 3

You create only one Connection. Try this:

private static enum Singleton {
    // Just one of me so constructor will be called once.
    Client;
    // The pool
    private PoolingHttpClientConnectionManager cm;

    // The constructor creates it - thus late
    private Singleton() {
      cm = new PoolingHttpClientConnectionManager();
      // Increase max total connection to 200
      cm.setMaxTotal(200);
      // Increase default max connection per route to 20
      cm.setDefaultMaxPerRoute(20);

    }

    public CloseableHttpClient get() {
      CloseableHttpClient threadSafeClient = HttpClients.custom()
              .setConnectionManager(cm)
              .build();
              return threadSafeClient;
    }

}

`