public class ExchangeClientConfig extends AbstractExchangeConfig {
	// in milliseconds
	private int connectTimeout = RemotingConstants.DEFAULT_CONNECT_TIMEOUT;
	private int reconnectInterval = RemotingConstants.DEFAULT_RECONNECT_INTERVAL;

	private short connections = 1;
	private boolean reversed = true;
	// enable channels Read/Write Splitting or not when connections greater than one
	private boolean enableReadWriteSplitting = false;
	/**
	 * write channels ratio when enableReadWriteSplitting is true, take writeConnections first if
	 * you don't specify writeConnections, writeRatio will be used
	 */
	private int writeRatio = 50;
	private short writeConnections;
	private boolean lazyConnection = false;

	public ChannelGroupMode mode() {
		if (connections == 1) {
			return ChannelGroupMode.simple;
		}
		if (enableReadWriteSplitting) {
			return ChannelGroupMode.readwrite;
		}
		return ChannelGroupMode.multiconns;
	}

	@Override
	public void check() {
		super.check();
		if (enableReadWriteSplitting) {
			if (connections <= 1) {
				throw new IllegalArgumentException(
						"connections must greater than one when enableReadWriteSplitting is true");
			}
			if (writeConnections == 0) {
				writeConnections = (short)(connections * (writeRatio / 100));
			}
			if (writeConnections == 0 || writeConnections == connections) {
				throw new IllegalArgumentException("writeConnections:" + writeConnections
						+ ", you should specify an appropriate writeConnections or writeRatio");
			}
		}
	}

	public short connections() {
		return connections;
	}

	public void connections(short connections) {
		this.connections = ParameterChecker.checkPositive(connections, "connections");
	}

	public short writeConnections() {
		return writeConnections;
	}

	public void writeConnections(short writeConnections) {
		this.writeConnections = ParameterChecker.checkPositive(writeConnections, "writeConnections");
	}

	public int reconnectInterval() {
		return reconnectInterval;
	}

	public void reconnectInterval(int reconnectInterval) {
		this.reconnectInterval = ParameterChecker.checkPositive(reconnectInterval, "reconnectInterval");
	}

	public int connectTimeout() {
		return connectTimeout;
	}

	public void connectTimeout(int connectTimeout) {
		this.connectTimeout = ParameterChecker.checkPositive(connectTimeout, "connectTimeout");
	}

	public boolean reversed() {
		return reversed;
	}

	public void reversed(boolean reversed) {
		this.reversed = reversed;
	}

	public boolean enableReadWriteSplitting() {
		return enableReadWriteSplitting;
	}

	public void enableReadWriteSplitting(boolean enableReadWriteSplitting) {
		this.enableReadWriteSplitting = enableReadWriteSplitting;
	}

	public int writeRatio() {
		return writeRatio;
	}

	public void writeRatio(int writeRatio) {
		if (writeRatio <= 0 || writeRatio > 100) {
			throw new IllegalArgumentException("writeRatio: " + writeRatio + " (expected: 0 < writeRatio <= 100)");
		}
		this.writeRatio = writeRatio;
	}

	public boolean isLazyConnection() {
		return lazyConnection;
	}

	public void setLazyConnection(boolean lazyConnection) {
		this.lazyConnection = lazyConnection;
	}
} public class ExchangeServerConfig extends AbstractExchangeConfig{

	@Override
	public void check() {
		super.check();
	}
	
} public enum ExceptionCode {
    BAD_PACKAGE,
    RESPONSE_TIMEOUT,
    WRITE_TIMEOUT,
    EXCHANGER_NOT_AVAILABLE,
    UNFINISHED_REQUEST,
    CHANNEL_WRITE_FAIL,
    DEFAULT,
    ;
} public class SailfishException extends Exception {

    /**  */
    private static final long serialVersionUID = 1L;
    private ExceptionCode     errorCode;

    public SailfishException(String message) {
        super(message);
    }
    
    public SailfishException(Throwable cause){
        super(cause);
    }

    public SailfishException(ExceptionCode errorCode, String message) {
        super(message(errorCode, message));
        this.errorCode = errorCode;
    }

    public SailfishException(String message, Throwable cause) {
        super(message, cause);
    }

    public SailfishException(ExceptionCode errorCode, String message, Throwable cause) {
        super(message(errorCode, message), cause);
        this.errorCode = errorCode;
    }

    private static String message(ExceptionCode errorCode, String message) {
        if(null == errorCode){
            errorCode = ExceptionCode.DEFAULT;
        }
        String prefix = "[errorCode:" + errorCode.toString() + "]";
        return StrUtils.isBlank(message) ? prefix : prefix + ", "+ message;
    }

    public ExceptionCode code() {
        return errorCode;
    }

    public RemoteSailfishException toRemoteException() {
        return new RemoteSailfishException(errorCode, getMessage(), getCause());
    }
    
    /**
     * exception for remote peer
     */
    class RemoteSailfishException extends SailfishException{
        private static final long serialVersionUID = 1L;
        public RemoteSailfishException(ExceptionCode errorCode, String message, Throwable cause) {
            super(errorCode, message, cause);
        }
    }
} public class SimpleExecutor implements Executor, Runnable {

	private static final Logger logger = LoggerFactory.getLogger(SimpleExecutor.class);
	
	private static final boolean preferGlobalEventExecutor;
	static{
		preferGlobalEventExecutor = SystemPropertyUtil.getBoolean("sailfish.simpleExector.preferGlobalEventExecutor", true);
		if(logger.isDebugEnabled()){
			logger.debug("-Dsailfish.simpleExector.preferGlobalEventExecutor: {}", preferGlobalEventExecutor);
		}
	}
	
	// produce FastThreadLocalThread(can benefit from FastThreadLocal(e.g. Recycler))
	private final DefaultThreadFactory THREADFACTORY = new DefaultThreadFactory("sailfish-simpleexecutor", true);

	public static final SimpleExecutor INSTANCE = new SimpleExecutor();
	private final LinkedList<Runnable> tasks = new LinkedList<>();
	private volatile boolean isRunning = false;

	private SimpleExecutor() { }

	@Override
	public void execute(Runnable task) {
		if(preferGlobalEventExecutor){
			GlobalEventExecutor.INSTANCE.execute(task);
			return;
		}
		synchronized (tasks) {
			tasks.add(task);
			if (!this.isRunning) {
				this.isRunning = true;
				newThread().start();
			}
		}
	}

	private Thread newThread() {
		return THREADFACTORY.newThread(this);
	}

	@Override
	public void run() {
		for (;;) {
			Runnable currentTask = null;
			synchronized (tasks) {
				currentTask = tasks.poll();
				if (null == currentTask) {
					this.isRunning = false;
					break;
				}
			}
			try {
				currentTask.run();
			} catch (Throwable cause) {
				logger.error("catch exception by SimpleExecutor", cause);
			}
		}
	}
} public interface ResponseFuture<T>{
    void putResponse(T resp, byte result, SailfishException cause);
    boolean isDone();
    void setCallback(ResponseCallback<T> callback, int timeout);
    T get() throws SailfishException, InterruptedException;
    T get(long timeout, TimeUnit unit) throws SailfishException, InterruptedException;
} public class ClientServerNormalRequestTestRequestProcessor implements RequestProcessor{
	public static final short OPCODE = 1;
	@Override
	public Executor executor() {
		return null;
	}

	@Override
	public short opcode() {
		return OPCODE;
	}

	@Override
	public void handleRequest(Request request, Output output) {
		Assert.assertNotNull(request.getRequestData());
        Assert.assertTrue(request.getRequestData().length > 0);
        Assert.assertArrayEquals(ClientServerTest.data, request.getRequestData());
        output.response(new Response(true, ClientServerTest.data));
	}

	@Override
	public void onRejectedExecutionException(Request request, Output output) {
	}
} public class ClientServerOnewayTestRequestProcessor implements RequestProcessor{
	public static final short OPCODE = 0;
	@Override
	public Executor executor() {
		return null;
	}

	@Override
	public short opcode() {
		return OPCODE;
	}

	@Override
	public void handleRequest(Request request, Output output) {
		ClientServerTest.RECORDS.get(Bytes.bytes2int(request.getRequestData())).countDown();
	}

	@Override
	public void onRejectedExecutionException(Request request, Output output) {
	}
} public class ClientServerTimeoutTestRequestProcessor implements RequestProcessor{
	public static final short OPCODE = 2;
	@Override
	public Executor executor() {
		return null;
	}

	@Override
	public short opcode() {
		return OPCODE;
	}

	@Override
	public void handleRequest(Request request, Output output) {
		//do nothing, request from remote peer will timeout
	}

	@Override
	public void onRejectedExecutionException(Request request, Output output) {
	}
} public interface RequestProcessor {
	Executor executor();

	short opcode();

	void handleRequest(Request request, Output output);

	void onRejectedExecutionException(Request request, Output output);

	interface Output {
		void response(Response response);
	}
} public class Bytes {
    public static int bytes2int(byte[] bytes) {
        if (null == bytes || bytes.length != 4) {
            throw new IllegalArgumentException();
        }
        return bytes[0] << 24 | bytes[1] << 16 | bytes[2] << 8 | bytes[3] << 0;
    }

    //high byte first
    public static byte[] int2bytes(int integer) {
        byte[] bytes = new byte[4];
        bytes[0] = (byte) (integer >> 24 & 0xFF);
        bytes[1] = (byte) (integer >> 16 & 0xFF);
        bytes[2] = (byte) (integer >> 8 & 0xFF);
        bytes[3] = (byte) (integer >> 0 & 0xFF);
        return bytes;
    }
}
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.BeforeClass;
import org.junit.Test;
import io.netty.util.CharsetUtil;
import sailfish.remoting.configuration.ExchangeClientConfig;
import sailfish.remoting.configuration.ExchangeServerConfig;
import sailfish.remoting.exceptions.ExceptionCode;
import sailfish.remoting.exceptions.SailfishException;
import sailfish.remoting.executor.SimpleExecutor;
import sailfish.remoting.future.ResponseFuture;
import sailfish.remoting.processors.ClientServerNormalRequestTestRequestProcessor;
import sailfish.remoting.processors.ClientServerOnewayTestRequestProcessor;
import sailfish.remoting.processors.ClientServerTimeoutTestRequestProcessor;
import sailfish.remoting.processors.RequestProcessor;
import sailfish.remoting.utils.Bytes;
/**
 *
 *	Copyright 2016-2016 spccold
 *
 *	Licensed under the Apache License, Version 2.0 (the "License");
 *	you may not use this file except in compliance with the License.
 *	You may obtain a copy of the License at
 *
 *   	http://www.apache.org/licenses/LICENSE-2.0
 *
 *	Unless required by applicable law or agreed to in writing, software
 *	distributed under the License is distributed on an "AS IS" BASIS,
 *	WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *	See the License for the specific language governing permissions and
 *	limitations under the License.
 *
 */
package sailfish.remoting;




/**
 * 
 * @author spccold
 * @version $Id: ClientServerTest.java, v 0.1 2016年10月26日 下午4:36:23 jileng Exp $
 */
public class ClientServerTest {
    public volatile static byte                               data[]                   = "hello sailfish!"
        .getBytes(CharsetUtil.UTF_8);
    private volatile static AtomicInteger                      ONEWAY_PAYLOAD_GENERATOR = new AtomicInteger(0);
    public static final Map<Integer, CountDownLatch> RECORDS                  = new HashMap<>();
    private static DefaultServer                     server;
    private static int                                originPort               = 13141;

    @BeforeClass
    public static void beforeClass() throws Exception {
        ExchangeServerConfig serverConfig = new ExchangeServerConfig();
        serverConfig.address(new Address("localhost", originPort));
        List<RequestProcessor> processors = new ArrayList<>(3);
        processors.add(new ClientServerOnewayTestRequestProcessor());
        processors.add(new ClientServerNormalRequestTestRequestProcessor());
        processors.add(new ClientServerTimeoutTestRequestProcessor());
        serverConfig.setRequestProcessors(processors);

        server = Exchanger.bind(serverConfig);
        server.start();
    }

    @AfterClass
    public static void afterClass() throws InterruptedException {
        if (null != server) {
        	server.close();
        }
    }

    private ExchangeClientConfig newBaseConfig(int port) {