package org.handwerkszeug.riak.pbc;

import static org.handwerkszeug.riak.util.Validation.notNull;

import java.net.SocketAddress;
import java.util.concurrent.Executors;

import org.handwerkszeug.riak.RiakAction;
import org.handwerkszeug.riak.RiakClient;
import org.jboss.netty.bootstrap.ClientBootstrap;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.ChannelStateEvent;
import org.jboss.netty.channel.SimpleChannelUpstreamHandler;
import org.jboss.netty.channel.socket.ClientSocketChannelFactory;
import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;

public class PbcRiakClient implements RiakClient<PbcRiakOperations> {

	final SocketAddress remoteHost;

	final ClientSocketChannelFactory channelFactory;

	public PbcRiakClient(SocketAddress remoteHost) {
		// TODO read from configuration.
		this.channelFactory = new NioClientSocketChannelFactory(
				Executors.newCachedThreadPool(),
				Executors.newCachedThreadPool());
		this.remoteHost = remoteHost;
	}

	@Override
	public void execute(final RiakAction<PbcRiakOperations> action) {
		notNull(action, "action");
		ClientBootstrap bootstrap = new ClientBootstrap(this.channelFactory);
		ChannelPipelineFactory pf = new PbcPipelineFactory() {
			@Override
			public ChannelPipeline getPipeline() throws Exception {
				ChannelPipeline pipeline = super.getPipeline();
				pipeline.addFirst("clientHandler",
						new SimpleChannelUpstreamHandler() {
							@Override
							public void channelConnected(
									ChannelHandlerContext ctx,
									ChannelStateEvent e) throws Exception {
								Channel c = e.getChannel();
								try {
									PbcRiakOperations op = new PbcRiakOperations(
											c);
									action.execute(op);
								} finally {

								}

							}
						});
				return pipeline;
			}
		};
		bootstrap.setPipelineFactory(pf);
		bootstrap.connect(this.remoteHost);
	}

	@Override
	public void dispose() {
		this.channelFactory.releaseExternalResources();
	}
}

@@ -312,49 +312,8 @@
 p;%0D%0A
-import org.jboss.netty.channel.Channel;%0D%0A
 impo
@@ -1259,16 +1259,73 @@
 ion%22);%0D%0A
+%09%09// TODO stress test and implement connection pooling.%0D%0A
 %09%09Client
@@ -1805,61 +1805,8 @@
  %7B%0D%0A
-%09%09%09%09%09%09%09%09Channel c = e.getChannel();%0D%0A%09%09%09%09%09%09%09%09try %7B%0D%0A%09
 %09%09%09%09
@@ -1854,16 +1854,17 @@
 rations(
+e
 %0D%0A%09%09%09%09%09%09
@@ -1871,15 +1871,25 @@
 %09%09%09%09
-%09c
+.getChannel()
 );%0D%0A
-%09
 %09%09%09%09
@@ -1917,44 +1917,8 @@
 );%0D%0A
-%09%09%09%09%09%09%09%09%7D finally %7B%0D%0A%0D%0A%09%09%09%09%09%09%09%09%7D%0D%0A%0D%0A
 %09%09%09%09

remove dead code