package org.msgpack.rpc.client;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.concurrent.Executors;

import org.jboss.netty.bootstrap.ClientBootstrap;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelFactory;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;
import org.msgpack.rpc.client.netty.RPCClientPipelineFactory;

public class TCPSocket {
	protected Address address;
	protected EventLoop loop;
	protected TCPTransport transport;

	// netty-specific
	protected ClientBootstrap bootstrap;
	protected ChannelFuture connectFuture;
	protected Channel channel;

	public TCPSocket(Address address, EventLoop loop, TCPTransport transport) {
		this.address = address;
		this.loop = loop;
		this.transport = transport;
		this.connectFuture = null;
		this.channel = null;
		this.bootstrap = loop.createBootstrap();
		this.bootstrap.setPipelineFactory(new RPCClientPipelineFactory(this));
	}
	
	public synchronized void tryConnect() throws Exception {
		if (connectFuture != null)
			throw new IOException("already connected");
		connectFuture = bootstrap.connect(new InetSocketAddress(address.getHost(), address.getPort()));
	}
	
	public synchronized void trySend(Object msg) throws Exception {
		if (connectFuture == null || channel == null)
			throw new IOException("not connected, but try send");
		channel.write(msg);
	}
	
	public synchronized void tryClose() {
		if (channel != null && channel.isOpen())
			channel.close().awaitUninterruptibly();
		address = null;
		transport = null;
		connectFuture = null;
		channel = null;
	}

	// callback
	public synchronized void onConnected() throws Exception {
		// connected, but onConnected() called
		if (channel != null)
			throw new IOException("already connected");
		// onConnected() called without tryConnect
		if (connectFuture == null)
			throw new IOException("tryConnect was not called");

		// set channel
		channel = connectFuture.awaitUninterruptibly().getChannel();
		if (connectFuture.isSuccess())
			transport.onConnected();
		else
			onConnectFailed();
	}

	// callback
	public synchronized void onConnectFailed() {
		if (transport != null)
			transport.onConnectFailed();
		tryClose();
	}

	// callback
	public synchronized void onMessageReceived(Object replyObject) throws Exception {
		if (transport != null)
			transport.onMessageReceived(replyObject);
	}
	
	// callback
	public synchronized void onClosed() {
		if (transport != null)
			transport.onClosed();
		tryClose();
	}
	
	// callback
	public synchronized void onFailed(Exception e) {
		if (transport != null)
			transport.onFailed(e);
		tryClose();
	}
}

@@ -484,16 +484,22 @@
 otected 
+final 
 Address 
@@ -518,16 +518,22 @@
 otected 
+final 
 EventLoo
@@ -551,16 +551,22 @@
 otected 
+final 
 TCPTrans
@@ -1597,46 +1597,8 @@
 ();%0A
-%09%09address = null;%0A%09%09transport = null;%0A
 %09%09co
@@ -2143,37 +2143,24 @@
 ack%0A%09public 
-synchronized 
 void onConne
@@ -2176,34 +2176,8 @@
 ) %7B%0A
-%09%09if (transport != null)%0A%09
 %09%09tr
@@ -2238,37 +2238,24 @@
 ack%0A%09public 
-synchronized 
 void onMessa
@@ -2308,34 +2308,8 @@
 n %7B%0A
-%09%09if (transport != null)%0A%09
 %09%09tr
@@ -2370,37 +2370,24 @@
 ack%0A%09public 
-synchronized 
 void onClose
@@ -2396,34 +2396,8 @@
 ) %7B%0A
-%09%09if (transport != null)%0A%09
 %09%09tr
@@ -2502,34 +2502,8 @@
 ) %7B%0A
-%09%09if (transport != null)%0A%09
 %09%09tr

add final to immutable variables, and remove synchronized from onMessageReceived()