In memory request handling on the server side with Jersey

Jersey ServerSide connector

It's possible to handle requests in memory with Jersey client

httpclient-requests

2013-08-14 22:17:38 INFO  MessageClient:57 - Time elapsed (get): 5ms
2013-08-14 22:17:38 INFO  MessageClient:57 - Time elapsed (get): 6ms
2013-08-14 22:17:38 INFO  MessageClient:57 - Time elapsed (get): 5ms
2013-08-14 22:17:38 INFO  MessageClient:57 - Time elapsed (get): 4ms
2013-08-14 22:17:38 INFO  MessageClient:57 - Time elapsed (get): 5ms
2013-08-14 22:17:38 INFO  MessageClient:57 - Time elapsed (get): 4ms
2013-08-14 22:17:39 INFO  MessageClient:57 - Time elapsed (get): 4ms
2013-08-14 22:17:39 INFO  MessageClient:57 - Time elapsed (get): 6ms
2013-08-14 22:17:39 INFO  MessageClient:57 - Time elapsed (get): 5ms
2013-08-14 22:17:39 INFO  MessageClient:57 - Time elapsed (get): 4ms

in-memory-requests

2013-08-14 20:52:10 INFO  MessageClient:57 - Time elapsed (get): 2ms
2013-08-14 20:52:10 INFO  MessageClient:57 - Time elapsed (get): 2ms
2013-08-14 20:52:10 INFO  MessageClient:57 - Time elapsed (get): 2ms
2013-08-14 20:52:10 INFO  MessageClient:57 - Time elapsed (get): 2ms
2013-08-14 20:52:10 INFO  MessageClient:57 - Time elapsed (get): 2ms
2013-08-14 20:52:10 INFO  MessageClient:57 - Time elapsed (get): 2ms
2013-08-14 20:52:10 INFO  MessageClient:57 - Time elapsed (get): 1ms
2013-08-14 20:52:10 INFO  MessageClient:57 - Time elapsed (get): 2ms
2013-08-14 20:52:10 INFO  MessageClient:57 - Time elapsed (get): 2ms
2013-08-14 20:52:10 INFO  MessageClient:57 - Time elapsed (get): 1ms

MessageClient.java

import java.util.List;

import javax.ws.rs.client.Client;
import javax.ws.rs.client.ClientBuilder;
import javax.ws.rs.client.Entity;
import javax.ws.rs.core.GenericType;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;

import org.apache.http.HttpHost;
import org.apache.http.conn.routing.HttpRoute;
import org.apache.http.impl.conn.PoolingClientConnectionManager;
import org.apache.log4j.Logger;
import org.glassfish.jersey.client.ClientConfig;
import org.glassfish.jersey.client.ClientProperties;
import org.glassfish.jersey.jackson.JacksonFeature;

public final class MessageClient {
  
  private static final MessageClient INSTANCE = new MessageClient();
  
  private final Client client;
  
  private static final Logger LOGGER = Logger.getLogger(MessageClient.class);

  private MessageClient() {
    ClientConfig clientConfig = new ClientConfig();
    clientConfig.property(ClientProperties.READ_TIMEOUT, 2000);
    clientConfig.property(ClientProperties.CONNECT_TIMEOUT, 500);
    
    PoolingClientConnectionManager connectionManager = new PoolingClientConnectionManager();
    connectionManager.setMaxTotal(100);
    connectionManager.setDefaultMaxPerRoute(20);
    
    // uncomment these two lines and comment out line 40 to use HTTP Client instead
//    clientConfig.property(ApacheClientProperties.CONNECTION_MANAGER, connectionManager);
//    ApacheConnector connector = new ApacheConnector(clientConfig);
    
    ServerSideConnector connector = ServerConnectorFactory.build();
    clientConfig.connector(connector);
    
    client = ClientBuilder.newClient(clientConfig);
    client.register(JacksonFeature.class);
  }
  
  public static List<Message> get(String name) {
    long start = System.currentTimeMillis();

    try {
      return INSTANCE.client.target("http://localhost:8080/myapp").path("messages/" + name)
          .request(MediaType.APPLICATION_JSON).get(new GenericType<List<Message>>(){});
    } finally {
      LOGGER.info("Time elapsed (get): " + (System.currentTimeMillis() - start) + "ms");
    }
  }
  
  public static void put(String name, List<Message> messages) {
    Response response = INSTANCE.client.target("http://localhost:8080/myapp").path("messages/" + name)
      .request(MediaType.APPLICATION_JSON).put(Entity.entity(messages, MediaType.APPLICATION_JSON));
    if (response.getStatus() != 200) {
      throw new RuntimeException("Response was " + response.getStatus());
    }
  }
}

MyApplication.java

import javax.ws.rs.ApplicationPath;

import org.glassfish.jersey.server.ResourceConfig;

@ApplicationPath("myapp")
public class MyApplication extends ResourceConfig {
  
  public MyApplication() {
    // other boostrapping options go here
    register(ServerConnectorProvider.class);
  }

}

ServerConnectorProvider.java

import javax.ws.rs.core.Context;
import javax.ws.rs.core.Feature;
import javax.ws.rs.core.FeatureContext;
import javax.ws.rs.ext.Provider;

import org.glassfish.jersey.server.ApplicationHandler;

/*
 * This is truly ugly, unrefined code.  I'm not going to say anything to defend otherwise.
 */
@Provider
public class ServerConnectorProvider implements Feature {
  
  private ApplicationHandler applicationHandler;
  
  public ServerConnectorProvider(@Context ApplicationHandler applicationHandler) {
    this.applicationHandler = applicationHandler;
  }
  
  @Override
  public boolean configure(FeatureContext context) {
    ServerConnectorFactory.init(applicationHandler);
    return true;
  }
  
  public static final class ServerConnectorFactory {
    
    private static ApplicationHandler applicationHandler;
    
    private static void init(ApplicationHandler applicationHandler) {
      ServerConnectorFactory.applicationHandler = applicationHandler;
    }
    
    private ServerConnectorFactory() {}
    
    public static ServerSideConnector build() {
      return new ServerSideConnector(applicationHandler);
    }
    
  }

}

ServerSideConnector.java

import java.io.IOException;
import java.io.OutputStream;
import java.io.StringWriter;
import java.net.URI;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

import javax.ws.rs.ProcessingException;
import javax.ws.rs.core.MultivaluedMap;
import javax.ws.rs.core.Response;
import javax.ws.rs.core.Response.ResponseBuilder;

import org.apache.commons.io.IOUtils;
import org.apache.commons.io.output.WriterOutputStream;
import org.apache.log4j.Logger;
import org.glassfish.jersey.client.ClientRequest;
import org.glassfish.jersey.client.ClientResponse;
import org.glassfish.jersey.client.spi.AsyncConnectorCallback;
import org.glassfish.jersey.client.spi.Connector;
import org.glassfish.jersey.message.internal.OutboundMessageContext.StreamProvider;
import org.glassfish.jersey.server.ApplicationHandler;
import org.glassfish.jersey.server.ContainerRequest;
import org.glassfish.jersey.server.ContainerResponse;

import com.google.common.util.concurrent.MoreExecutors;

public class ServerSideConnector implements Connector {

  private ApplicationHandler applicationHandler;

  private Logger logger = Logger.getLogger(getClass());

  public ServerSideConnector(ApplicationHandler applicationHandler) {
    this.applicationHandler = applicationHandler;
  }

  @Override
  public ClientResponse apply(ClientRequest request) throws ProcessingException {
    ContainerRequest containerRequest = buildContainerRequestFromClientRequest(request);
    Future<ContainerResponse> future = applicationHandler.apply(containerRequest);
    ContainerResponse containerResponse = null;
    try {
      containerResponse = future.get();
    } catch (InterruptedException | ExecutionException e) {
      logger.error(e);
      throw new ProcessingException(e);
    }
    Response response = buildResponseFromContainerResponse(containerResponse);
    ClientResponse clientResponse = new ClientResponse(request, response);
    return clientResponse;
  }

  private Response buildResponseFromContainerResponse(ContainerResponse containerResponse) {
    ResponseBuilder responseBuilder = Response.status(containerResponse.getStatusInfo())
        .entity(containerResponse.getEntity()).lastModified(containerResponse.getLastModified())
        .type(containerResponse.getMediaType());

    for (String key : containerResponse.getHeaders().keySet()) {
      // For some bizarre reason, the Content-Type header can get written out
      // twice and cause an exception to be thrown
      if (!"Content-Type".equals(key)) {
        List<Object> values = containerResponse.getHeaders().get(key);
        for (Object value : values) {
          responseBuilder.header(key, value);
        }
      }
    }

    Response response = responseBuilder.build();
    return response;
  }

  private ContainerRequest buildContainerRequestFromClientRequest(ClientRequest request) {
    URI uri = request.getUri();
    String method = request.getMethod();
    // you'll need to configure your endpoint's base uri somewhere for your client anyway.
    // in this case it's inlined
    ContainerRequest containerRequest = new ContainerRequest(URI.create("http://localhost:8080/myapp"), uri, method,
        null, null);

    // copy the headers
    MultivaluedMap<String, Object> headers = request.getHeaders();
    for (String key : headers.keySet()) {
      List<Object> values = headers.get(key);
      for (Object value : values) {
        containerRequest.header(key, value);
      }
    }

    // copy the request entity
    if (request.getEntity() != null) {
      try {
        final StringWriter writer = new StringWriter();
        final OutputStream output = new WriterOutputStream(writer);
        request.setStreamProvider(new StreamProvider() {

          @Override
          public OutputStream getOutputStream(int contentLength) throws IOException {
            return output;
          }
        });
        
        // you have to set the stream provider before calling this method
        request.writeEntity();
        containerRequest.setEntityStream(IOUtils.toInputStream(writer.toString()));
      } catch (IOException e) {
        logger.error(e);
        throw new RuntimeException(e);
      }
    }

    return containerRequest;
  }

  /* (non-Javadoc)
   * This is a total ripoff of the implementation in ApacheConnector#apply(request, callback)
   * @see org.glassfish.jersey.client.spi.Connector#apply(org.glassfish.jersey.client.ClientRequest, org.glassfish.jersey.client.spi.AsyncConnectorCallback)
   */
  @Override
  public Future<?> apply(final ClientRequest request, final AsyncConnectorCallback callback) {
    return MoreExecutors.sameThreadExecutor().submit(new Runnable() {
      @Override
      public void run() {
        try {
          callback.response(apply(request));
        } catch (ProcessingException ex) {
          callback.failure(ex);
        } catch (Throwable t) {
          callback.failure(t);
        }
      }
    });
  }

  @Override
  public String getName() {
    return "ServerSideConnector";
  }

  @Override
  public void close() {
  }

}

ServerSideConnectorApply.java

private ApplicationHandler applicationHandler;

public ServerSideConnector(ApplicationHandler applicationHandler) {
  this.applicationHandler = applicationHandler;
}

@Override
public ClientResponse apply(ClientRequest request) throws ProcessingException {
  ContainerRequest containerRequest = buildContainerRequestFromClientRequest(request);
  ... = applicationHandler.apply(containerRequest);
  ClientResponse clientResponse = ...
  return clientResponse;
}

ServerSideConnectorApplyFull.java

@Override
public ClientResponse apply(ClientRequest request) throws ProcessingException {
  ContainerRequest containerRequest = buildContainerRequestFromClientRequest(request);
  Future<ContainerResponse> future = applicationHandler.apply(containerRequest);
  ContainerResponse containerResponse = null;
  try {
    containerResponse = future.get();
  } catch (InterruptedException | ExecutionException e) {
    logger.error(e);
    throw new ProcessingException(e);
  }
  Response response = buildResponseFromContainerResponse(containerResponse);
  ClientResponse clientResponse = new ClientResponse(request, response);
  return clientResponse;
}

ServerSideConnectorBuildRequest.java

private ContainerRequest buildContainerRequestFromClientRequest(ClientRequest request) {
  URI uri = request.getUri();
  String method = request.getMethod();
  // you'll need to configure your endpoint's base uri somewhere for your client anyway.
  // in this case it's inlined
  ContainerRequest containerRequest = new ContainerRequest(URI.create("http://localhost:8080/myapp"), uri, method,
      null, null);

  // copy the headers
  MultivaluedMap<String, Object> headers = request.getHeaders();
  for (String key : headers.keySet()) {
    List<Object> values = headers.get(key);
    for (Object value : values) {
      containerRequest.header(key, value);
    }
  }

  // copy the request entity
  if (request.getEntity() != null) {
    try {
      final StringWriter writer = new StringWriter();
      final OutputStream output = new WriterOutputStream(writer);
      request.setStreamProvider(new StreamProvider() {

        @Override
        public OutputStream getOutputStream(int contentLength) throws IOException {
          return output;
        }
      });
      
      // you have to set the stream provider before calling this method
      request.writeEntity();
      containerRequest.setEntityStream(IOUtils.toInputStream(writer.toString()));
    } catch (IOException e) {
      logger.error(e);
      throw new RuntimeException(e);
    }
  }

  return containerRequest;
}

ServerSideConnectorBuildResponse.java

private Response buildResponseFromContainerResponse(ContainerResponse containerResponse) {
  ResponseBuilder responseBuilder = Response.status(containerResponse.getStatusInfo())
      .entity(containerResponse.getEntity()).lastModified(containerResponse.getLastModified())
      .type(containerResponse.getMediaType());

  for (String key : containerResponse.getHeaders().keySet()) {
    // For some bizarre reason, the Content-Type header can get written out
    // twice and cause an exception to be thrown
    if (!"Content-Type".equals(key)) {
      List<Object> values = containerResponse.getHeaders().get(key);
      for (Object value : values) {
        responseBuilder.header(key, value);
      }
    }
  }

  Response response = responseBuilder.build();
  return response;
}

ServerSideConnectorConstructor.java

import org.apache.log4j.Logger;
import org.glassfish.jersey.client.spi.Connector;
import org.glassfish.jersey.server.ApplicationHandler;


public class ServerSideConnector implements Connector {

  private ApplicationHandler applicationHandler;

  private Logger logger = Logger.getLogger(getClass());

  public ServerSideConnector(ApplicationHandler applicationHandler) {
    this.applicationHandler = applicationHandler;
  }
  
  ...
  
}

ServerSideConnectorSendRequest.java

@Override
public ClientResponse apply(ClientRequest request) throws ProcessingException {
  ContainerRequest containerRequest = buildContainerRequestFromClientRequest(request);
  Future<ContainerResponse> future = applicationHandler.apply(containerRequest);
  ContainerResponse containerResponse = null;
  try {
    containerResponse = future.get();
  } catch (InterruptedException | ExecutionException e) {
    logger.error(e);
    throw new ProcessingException(e);
  }
  
  ClientResponse clientResponse = ...
  return clientResponse;
}