gistfile1.java

import java.net.InetSocketAddress;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

import org.jboss.netty.bootstrap.ServerBootstrap;
import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;
import org.jboss.netty.channel.*;
import static org.jboss.netty.channel.Channels.pipeline;
import org.jboss.netty.handler.execution.OrderedMemoryAwareThreadPoolExecutor;
import org.jboss.netty.handler.execution.ExecutionHandler;
import org.jboss.netty.handler.codec.http.*;
import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.buffer.ChannelBuffers;

public class HttpServer
{
	public static void main(String[] args)
	{
		ServerBootstrap bootstrap = new ServerBootstrap(
				new NioServerSocketChannelFactory(
						Executors.newCachedThreadPool(),
						Executors.newCachedThreadPool()));

		OrderedMemoryAwareThreadPoolExecutor pipelineExecutor = new OrderedMemoryAwareThreadPoolExecutor(1000, 0, 0);
		ExecutionHandler eh = new ExecutionHandler(pipelineExecutor);

		bootstrap.setPipelineFactory(new HttpServerPipelineFactory(eh));
		bootstrap.bind(new InetSocketAddress(8080));
	}
}

class HttpServerPipelineFactory implements ChannelPipelineFactory
{
	private ExecutionHandler executionHandler;

	public HttpServerPipelineFactory(ExecutionHandler executionHandler)
	{
		this.executionHandler = executionHandler;
	}

	public ChannelPipeline getPipeline() throws Exception
	{
		ChannelPipeline pipeline = pipeline();
		pipeline.addLast("decoder", new HttpRequestDecoder());
		pipeline.addLast("encoder", new HttpResponseEncoder());
		pipeline.addLast("pipelineExecutor", executionHandler);
		pipeline.addLast("handler", new HttpRequestHandler());
		return pipeline;
	}
}

@ChannelPipelineCoverage("one")
class HttpRequestHandler extends SimpleChannelUpstreamHandler
{
	private static AtomicInteger count = new AtomicInteger(0);
	private volatile HttpRequest request;
	private volatile boolean readingChunks;
	private final StringBuilder responseContent = new StringBuilder();

	@Override
	public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception
	{
		HttpRequest request = this.request = (HttpRequest) e.getMessage();

		Thread.sleep(1000);
		responseContent.append("Sleeping for 1000 ms\r\n");

		writeResponse(e);
	}

	private void writeResponse(MessageEvent e)
	{
		ChannelBuffer buf = ChannelBuffers.copiedBuffer(responseContent.toString(), "UTF-8");
		responseContent.setLength(0);

		HttpResponse response = new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK);
		response.setContent(buf);
		response.setHeader(HttpHeaders.Names.CONTENT_TYPE, "text/plain; charset=UTF-8");
		response.setHeader(HttpHeaders.Names.CONTENT_LENGTH, String.valueOf(buf.readableBytes()));
		ChannelFuture future = e.getChannel().write(response);
		future.addListener(ChannelFutureListener.CLOSE);
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e)
			throws Exception
	{
		e.getCause().printStackTrace();
		e.getChannel().close();
	}
}