public class DefaultServer implements Server {
	
	private volatile boolean isClosed = false;
	private final ExchangeServerConfig config;
	private final MsgHandler<Protocol> msgHandler;
	private Channel channel;

	public DefaultServer(ExchangeServerConfig config) {
		this.config = ParameterChecker.checkNotNull(config, "ExchangeServerConfig");
		this.msgHandler = new DefaultMsgHandler(config.getRequestProcessors());
	}

	public void start() throws SailfishException {
		ServerBootstrap boot = newServerBootstrap();
		EventLoopGroup accept = NettyPlatformIndependent.newEventLoopGroup(1,
				new DefaultThreadFactory(RemotingConstants.SERVER_ACCEPT_THREADNAME));
		if (null != config.getEventLoopGroup()) {
			boot.group(accept, config.getEventLoopGroup());
		} else {
			boot.group(accept, ServerEventGroup.INSTANCE.getLoopGroup());
		}
		final EventExecutorGroup executor = (null != config.getEventExecutorGroup() ? config.getEventExecutorGroup()
				: ServerEventGroup.INSTANCE.getExecutorGroup());
		boot.localAddress(config.address().host(), config.address().port());
		boot.childHandler(new ChannelInitializer<SocketChannel>() {
			@Override
			protected void initChannel(SocketChannel ch) throws Exception {
				ChannelPipeline pipeline = ch.pipeline();
				ch.attr(ChannelAttrKeys.OneTime.idleTimeout).set(config.idleTimeout());
				ch.attr(ChannelAttrKeys.maxIdleTimeout).set(config.maxIdleTimeout());
				ch.attr(ChannelAttrKeys.exchangeServer).set(DefaultServer.this);
				pipeline.addLast(executor, 
						RemotingEncoder.INSTANCE, 
						new RemotingDecoder(), 
						new IdleStateHandler(config.idleTimeout(), 0, 0), 
						HeartbeatChannelHandler.INSTANCE,
						NegotiateChannelHandler.INSTANCE,
						ConcreteRequestHandler.INSTANCE);
			}
		});
		try {
			channel = boot.bind().syncUninterruptibly().channel();
		} catch (Throwable cause) {
			throw new SailfishException(cause);
		}
	}

	private ServerBootstrap newServerBootstrap() {
		ServerBootstrap serverBoot = new ServerBootstrap();
		serverBoot.channel(NettyPlatformIndependent.serverChannelClass());
		// connections wait for accept
		serverBoot.option(ChannelOption.SO_BACKLOG, 1024);
		serverBoot.option(ChannelOption.SO_REUSEADDR, true);
		// replace by heart beat
		serverBoot.childOption(ChannelOption.SO_KEEPALIVE, false);
		serverBoot.childOption(ChannelOption.TCP_NODELAY, true);
		serverBoot.childOption(ChannelOption.SO_SNDBUF, 32 * 1024);
		serverBoot.childOption(ChannelOption.SO_RCVBUF, 32 * 1024);
		// temporary settings, need more tests
		serverBoot.childOption(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark(8 * 1024, 32 * 1024));
		serverBoot.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
		//default is true, reduce thread context switching
		serverBoot.childOption(ChannelOption.SINGLE_EVENTEXECUTOR_PER_GROUP, true);
		return serverBoot;
	}

	@Override
	public void close() {
		close(0);
	}

	@Override
	public void close(int timeout) {
		if(isClosed()){
			return;
		}
		synchronized (this) {
			if (isClosed())
				return;
			ChannelUtil.closeChannel(channel);
		}
	}

	@Override
	public boolean isClosed() {
		return this.isClosed;
	}

	public MsgHandler<Protocol> getMsgHandler() {
		return msgHandler;
	}

	@Override
	public Collection<ExchangeChannelGroup> listChannelGroups() {
		return NegotiateChannelHandler.uuid2ChannelGroup.values();
	}
} public interface ExchangeChannelGroup extends Endpoint, MessageExchangePattern{
	/**
	 * like {@link Channel#id()}, Returns the globally unique identifier of this {@link ExchangeChannelGroup}.
	 */
	UUID id();

	/**
	 * Return {@code true} if this {@link ExchangeChannelGroup} is available, this means that the {@link ExchangeChannelGroup}
	 * can receive bytes from remote peer or write bytes to remote peer
	 */
	boolean isAvailable();

	/**
     * Returns one of the {@link ExchangeChannel}s managed by this {@link ExchangeChannelGroup}.
     */
    ExchangeChannel next() throws SailfishException;
    
    /**
     * Return the {@link MsgHandler} of this {@link ExchangeChannelGroup} which used for process {@link ResponseProtocol} 
     * sent by the {@link ExchangeChannelGroup}
     */
    MsgHandler<Protocol> getMsgHander();
    
    /**
     * Return the {@link Tracer} of this {@link ExchangeChannelGroup} which used for trace {@link ResponseProtocol}
     * sent by the {@link ExchangeChannelGroup}
     * @return
     */
    Tracer getTracer();
} public final class ReadWriteServerExchangeChannelGroup extends ReferenceCountedServerExchangeChannelGroup{
	
	private static final Logger logger = LoggerFactory.getLogger(ReadWriteServerExchangeChannelGroup.class);
	
	private final ServerExchangeChannelGroup readGroup;
	private final ServerExchangeChannelGroup writeGroup;
	
	private final MsgHandler<Protocol> msgHandler;
	private final Tracer tracer;
	
	public ReadWriteServerExchangeChannelGroup(MsgHandler<Protocol> msgHandler, UUID id, int connctions, int writeConnections) {
		super(id);

		this.msgHandler = msgHandler;
		this.tracer = new Tracer();
		//read to write, write to read
		this.writeGroup = new ServerExchangeChannelGroup(tracer, msgHandler, id, connctions - writeConnections);
		this.readGroup = new ServerExchangeChannelGroup(tracer, msgHandler, id, writeConnections);
	}
	
	@Override
	public ExchangeChannel next() throws SailfishException {
		try {
			return writeGroup.next();
		} catch (SailfishException cause) {
			logger.warn("writeGroup not available, try to choose readGroup", cause);
		}
		//try readGroup if writeGroup no available
		return readGroup.next();
	}

	@Override
	public boolean isAvailable() {
		if (writeGroup.isAvailable()) {
			return true;
		}
		return readGroup.isAvailable();
	}

	@Override
	public void close(int timeout) {
		throw new UnsupportedOperationException("close with timeout: "+timeout);
	}

	@Override
	public MsgHandler<Protocol> getMsgHander() {
		return msgHandler;
	}

	@Override
	public Tracer getTracer() {
		return tracer;
	}

	@Override
	public void addChild(ExchangeChannel channel, NegotiateConfig config) {
		//read to write, write to read
		if(config.isRead()){
			writeGroup.addChild(channel, config);
		}else if(config.isWrite()){
			readGroup.addChild(channel, config);
		}
	}
} public abstract class ReferenceCountedServerExchangeChannelGroup extends AbstractExchangeChannelGroup {
	
	private static final AtomicIntegerFieldUpdater<ReferenceCountedServerExchangeChannelGroup> refCntUpdater;

	static {
		AtomicIntegerFieldUpdater<ReferenceCountedServerExchangeChannelGroup> updater = PlatformDependent
				.newAtomicIntegerFieldUpdater(ReferenceCountedServerExchangeChannelGroup.class, "refCnt");
		if (updater == null) {
			updater = AtomicIntegerFieldUpdater.newUpdater(ReferenceCountedServerExchangeChannelGroup.class, "refCnt");
		}
		refCntUpdater = updater;
	}

	private volatile int refCnt = 0;

	protected ReferenceCountedServerExchangeChannelGroup(UUID id) {
		super(id);
	}

	public ReferenceCountedServerExchangeChannelGroup retain() {
		return retain0(1);
	}

	public ReferenceCountedServerExchangeChannelGroup retain(int increment) {
		return retain0(checkPositive(increment, "increment"));
	}

	private ReferenceCountedServerExchangeChannelGroup retain0(int increment) {
		for (;;) {
			int refCnt = this.refCnt;
			final int nextCnt = refCnt + increment;
			if (refCntUpdater.compareAndSet(this, refCnt, nextCnt)) {
				break;
			}
		}
		return this;
	}
	
	public boolean release(){
		return release0(1);
	}
	
	public boolean release(int decrement) {
		return release0(checkPositive(decrement, "decrement"));
	}

	private boolean release0(int decrement) {
		for (;;) {
			int refCnt = this.refCnt;
			if (refCnt < decrement) {
				throw new IllegalReferenceCountException(refCnt, -decrement);
			}

			if (refCntUpdater.compareAndSet(this, refCnt, refCnt - decrement)) {
				if (refCnt == decrement) {
					deallocate();
					return true;
				}
				return false;
			}
		}
	}

	private void deallocate(){
		NegotiateChannelHandler.uuid2ChannelGroup.remove(id().toString());
	}
	
	public abstract void addChild(ExchangeChannel channel, NegotiateConfig config);
} public final class ServerExchangeChannel extends AbstractExchangeChannel {

	public ServerExchangeChannel(ExchangeChannelGroup parent, Channel channel) {
		super(parent);
		this.channel = channel;
	}

	@Override
	public Channel update(Channel newChannel) {
		throw new UnsupportedOperationException("update");
	}

	@Override
	public Channel doConnect() throws SailfishException {
		throw new UnsupportedOperationException("doConnect");
	}

	@Override
	public void recover() {
		throw new UnsupportedOperationException("recover");
	}
	
	@Override
	public void close(int timeout) {
		throw new UnsupportedOperationException("close with timeout: "+timeout);
	}

	@Override
	public boolean isAvailable() {
		return super.isAvailable() && super.isWritable();
	}

	@Override
	public MsgHandler<Protocol> getMsgHander() {
		return parent().getMsgHander();
	}

	@Override
	public Tracer getTracer() {
		return parent().getTracer();
	}
} public final class ServerExchangeChannelGroup extends ReferenceCountedServerExchangeChannelGroup{
	
	private final ExchangeChannel[] children;
	private final ExchangeChannel[] deadChildren;
	private final ExchangeChannelChooserFactory.ExchangeChannelChooser chooser;
	
	private final MsgHandler<Protocol> msgHandler;
	private final Tracer tracer;
	
	public ServerExchangeChannelGroup(MsgHandler<Protocol> msgHandler, UUID id, int connections){
		this(new Tracer(), msgHandler, id, connections);
	}
	
	public ServerExchangeChannelGroup(Tracer tracer, MsgHandler<Protocol> msgHandler, UUID id, int connections) {
		super(id);
		
		this.msgHandler = msgHandler;
		this.tracer = tracer;
		
		children = new ExchangeChannel[connections];
		deadChildren = new ExchangeChannel[connections];
		
		chooser = DefaultExchangeChannelChooserFactory.INSTANCE.newChooser(children, deadChildren);
	}

	@Override
	public ExchangeChannel next() throws SailfishException {
		return chooser.next();
	}

	@Override
	public boolean isAvailable() {
		if(children.length == 1){//one connection check
			return (null != children[0] && children[0].isAvailable());
		}
		
		// can hit most of the time
		if (deadChildren[0] == null || deadChildren[0].isAvailable()) {
			return true;
		}

		for (int i = 1; i < children.length; i++) {
			if (deadChildren[i] == null || deadChildren[i].isAvailable()) {
				return true;
			}
		}
		return false;
	}

	@Override
	public void close(int timeout) {
		throw new UnsupportedOperationException("close with timeout: "+timeout);
	}

	@Override
	public MsgHandler<Protocol> getMsgHander() {
		return msgHandler;
	}

	@Override
	public Tracer getTracer() {
		return tracer;
	}
	
	@Override
	public void addChild(ExchangeChannel channel, NegotiateConfig config) {
		children[config.index()] = channel;
	}
} public class NegotiateConfig {
	private byte idleTimeout;
	private byte maxIdleTimeout;

	private final UUID uuid;
	private final byte type;
	private final short connections;
	private final short writeConnections;
	private final boolean reversed;

	private short index;

	public NegotiateConfig(byte idleTimeout, byte maxIdleTimeout, UUID uuid, byte type, short connections,
			short writeConnections, short index, boolean reversed) {
		this.idleTimeout = idleTimeout;
		this.maxIdleTimeout = maxIdleTimeout;

		this.uuid = uuid;
		this.type = type;
		this.connections = connections;
		this.writeConnections = writeConnections;
		this.reversed = reversed;

		this.index = index;
	}

	public byte idleTimeout() {
		return this.idleTimeout;
	}

	public byte maxIdleTimeout() {
		return this.maxIdleTimeout;
	}

	public UUID uuid() {
		return this.uuid;
	}

	public byte type() {
		return this.type;
	}

	public short connections() {
		return this.connections;
	}

	public short writeConnections() {
		return this.writeConnections;
	}

	public NegotiateConfig index(short index) {
		this.index = index;
		return this;
	}

	public short index() {
		return this.index;
	}

	public boolean reversed() {
		return this.reversed;
	}

	public boolean isRead() {
		return ChannelType.read.code() == type;
	}

	public boolean isWrite() {
		return ChannelType.write.code() == type;
	}

	public boolean isReadWrite(){
		return ChannelType.readwrite.code() == type;
	}

	public void reverseIndex(){
		if(!this.reversed){
			return;
		}
		if(isReadWrite()){
			this.index = (short)ArrayUtils.reverseArrayIndex(this.connections, this.index);
		}else if(isRead()){
			int readConnections = this.connections - this.writeConnections;
			this.index = (short) ArrayUtils.reverseArrayIndex(readConnections, index);
		}else if(isWrite()){
			this.index = (short) ArrayUtils.reverseArrayIndex(this.writeConnections, index);
		}
	}
	
	public NegotiateConfig deepCopy() {
		return new NegotiateConfig(idleTimeout, maxIdleTimeout, uuid, type, connections, 
				writeConnections, index, reversed);
	}

	public RequestProtocol toNegotiateRequest() throws IOException {
		int size = 1 + 1 + 16 + 1 + 2 + 2 + 2 + 1;
		try (ByteArrayOutputStream baos = new ByteArrayOutputStream(size);
				DataOutputStream dos = new DataOutputStream(baos);) {
			dos.writeByte(this.idleTimeout);
			dos.writeByte(this.maxIdleTimeout);
			dos.writeLong(this.uuid.getMostSignificantBits());
			dos.writeLong(this.uuid.getLeastSignificantBits());
			dos.writeByte(this.type);
			dos.writeShort(this.connections);
			dos.writeShort(this.writeConnections);
			dos.writeShort(this.index);
			dos.writeBoolean(this.reversed);
			return RequestProtocol.newHeartbeat().opcode(Opcode.HEARTBEAT_WITH_NEGOTIATE).body(baos.toByteArray());
		}
	}

	public static NegotiateConfig fromNegotiate(byte[] negotiateData) throws IOException {
		try (DataInputStream dis = new DataInputStream(new ByteArrayInputStream(negotiateData))) {
			return new NegotiateConfig(dis.readByte(), dis.readByte(), new UUID(dis.readLong(), dis.readLong()),
					dis.readByte(), dis.readShort(), dis.readShort(), dis.readShort(), dis.readBoolean());
		}
	}
} public interface ChannelAttrKeys {
	AttributeKey<HighPerformanceChannelWriter> highPerformanceWriter = AttributeKey.valueOf("sailfish.highPerformanceWriter");
	
	// for idle handle and heart beat
	AttributeKey<Byte> maxIdleTimeout = AttributeKey.valueOf("sailfish.maxIdleTimeout");
	AttributeKey<Long> lastReadTimeMillis = AttributeKey.valueOf("sailfish.lastReadTimeMillis");

	// side
	AttributeKey<Boolean> clientSide = AttributeKey.valueOf("sailfish.side");

	AttributeKey<ExchangeChannelGroup> channelGroup = AttributeKey.valueOf("sailfish.channelGroup");
	AttributeKey<DefaultServer> exchangeServer = AttributeKey.valueOf("sailfish.exchangeServer");
	AttributeKey<String> uuidStr = AttributeKey.valueOf("sailfish.uuidStr");

	interface OneTime {
		// for idle handle
		AttributeKey<Byte> idleTimeout = AttributeKey.valueOf("sailfish.idleTimeout");
		AttributeKey<CountDownLatch> awaitNegotiate = AttributeKey.valueOf("sailfish.awaitNegotiate");

		AttributeKey<NegotiateConfig> channelConfig = AttributeKey.valueOf("sailfish.channelConfig");
	}
} interface OneTime {
	// for idle handle
	AttributeKey<Byte> idleTimeout = AttributeKey.valueOf("sailfish.idleTimeout");
	AttributeKey<CountDownLatch> awaitNegotiate = AttributeKey.valueOf("sailfish.awaitNegotiate");

	AttributeKey<NegotiateConfig> channelConfig = AttributeKey.valueOf("sailfish.channelConfig");
} public interface Opcode {
    short HEARTBEAT_WITH_NEGOTIATE = 1;
} public interface Protocol {
    /**
     * request or response direction
     */
    boolean request();
    
    /**
     * heartbeat request/response
     */
    boolean heartbeat();

    /**
     * serialize bytes data to channel
     */
    void serialize(ByteBuf output) throws SailfishException;
    
    /**
     * deserialize bytes data from channel
     */
    void deserialize(ByteBuf input, int totalLength) throws SailfishException;
} public class RequestProtocol implements Protocol {
	
	private static final Recycler<RequestProtocol> RECYCLER = new Recycler<RequestProtocol>(){
		@Override
		protected RequestProtocol newObject(Recycler.Handle<RequestProtocol> handle) {
			return new RequestProtocol(handle);
		}
	};
	
	private static final int HEADER_LENGTH = 8;
	public static final int REQUEST_FLAG = 0x80;
	private static final int ONEWAY_FLAG = 0x40;
	private static final int HEARTBEAT_FLAG = 0x20;

	
	public static RequestProtocol newInstance(){
		return RECYCLER.get();
	}
	
	private final Recycler.Handle<RequestProtocol> handle;
	
	// request direction
	private boolean heartbeat;
	private boolean oneway;
	private byte serializeType = SerializeType.NON_SERIALIZE;

	private int packetId;
	private short opcode;

	private byte compressType = CompressType.NON_COMPRESS;
	private byte langType = LangType.JAVA;

	private byte[] body;

	
	private RequestProtocol(Recycler.Handle<RequestProtocol> handle){
		this.handle = handle;
	}
	
	public void recycle(){
		if(null == handle){//some objects don't need recycle
			return;
		}
		heartbeat = false;
		oneway = false;
		serializeType = SerializeType.NON_SERIALIZE;
		packetId = 0;
		opcode = 0;
		compressType = CompressType.NON_COMPRESS;
		langType = LangType.JAVA;
		body = null;
		handle.recycle(this);
	}
	
	@Override
	public void serialize(ByteBuf output) throws SailfishException {
		try {
			// write magic first
			output.writeShort(RemotingConstants.SAILFISH_MAGIC);
			// write package length(not contain current length field(4 bytes))
			output.writeInt(HEADER_LENGTH + bodyLength());

			byte compactByte = (byte) REQUEST_FLAG;
			if (this.oneway) {
				compactByte = (byte) (compactByte | ONEWAY_FLAG);
			}
			if (this.heartbeat) {
				compactByte = (byte) (compactByte | HEARTBEAT_FLAG);
			}

			output.writeByte(compactByte | serializeType);

			output.writeInt(packetId);
			output.writeShort(opcode);

			output.writeByte(compressType << 4 | langType);

			if (bodyLength() != 0) {
				output.writeBytes(this.body);
			}
		} catch (Throwable cause) {
			throw new SailfishException(cause);
		}finally {
			recycle();
		}
	}

	@Override
	public void deserialize(ByteBuf input, int totalLength) throws SailfishException {
		try {
			byte compactByte = input.readByte();
			this.oneway = ((compactByte & ONEWAY_FLAG) != 0);
			this.heartbeat = ((compactByte & HEARTBEAT_FLAG) != 0);
			this.serializeType = (byte) (compactByte & 0x1F);

			this.packetId = input.readInt();
			this.opcode = input.readShort();

			byte tmp = input.readByte();
			this.compressType = (byte) (tmp >> 4 & 0xF);
			this.langType = (byte) (tmp >> 0 & 0xF);

			// read body
			int bodyLength = totalLength - HEADER_LENGTH;
			if (bodyLength > 0) {
				this.body = new byte[bodyLength];
				input.readBytes(this.body);
			}
		} catch (Throwable cause) {
			throw new SailfishException(cause);
		}
	}

	public RequestProtocol heartbeat(boolean heartbeat) {
		this.heartbeat = heartbeat;
		return this;
	}

	public boolean oneway() {
		return oneway;
	}

	public RequestProtocol oneway(boolean oneway) {
		this.oneway = oneway;
		return this;
	}

	public byte serializeType() {
		return serializeType;
	}

	public RequestProtocol serializeType(byte serializeType) {
		this.serializeType = ProtocolParameterChecker.checkSerializeType(serializeType);
		return this;
	}

	public int packetId() {
		return packetId;
	}

	public RequestProtocol packetId(int packetId) {
		this.packetId = packetId;
		return this;
	}

	public short opcode() {
		return opcode;
	}

	public RequestProtocol opcode(short opcode) {
		this.opcode = opcode;
		return this;
	}

	public byte compressType() {
		return compressType;
	}

	public RequestProtocol compressType(byte compressType) {
		this.compressType = ProtocolParameterChecker.checkCompressType(compressType);
		return this;
	}

	public byte langType() {
		return langType;
	}

	public RequestProtocol langType(byte langType) {
		this.langType = ProtocolParameterChecker.checkLangType(langType);
		return this;
	}

	public byte[] body() {
		return body;
	}

	public RequestProtocol body(byte[] body) {
		this.body = body;
		return this;
	}

	private int bodyLength() {
		if (null == body) {
			return 0;
		}
		return body.length;
	}

	@Override
	public boolean request() {
		return true;
	}

	public boolean heartbeat() {
		return this.heartbeat;
	}

	public Request toRequest(){
		Request request = new Request(oneway, serializeType, compressType, body, langType);
		return request;
	}
	
	@Override
	public String toString() {
		return "RequestProtocol [heartbeat=" + heartbeat + ", oneway=" + oneway + ", serializeType=" + serializeType
				+ ", packetId=" + packetId + ", opcode=" + opcode + ", compressType=" + compressType + ", langType="
				+ langType + ", body=" + Arrays.toString(body) + "]";
	}

	public static RequestProtocol newRequest(RequestControl requestControl) {
		RequestProtocol protocol = RequestProtocol.newInstance();
		protocol.packetId(PacketIdGenerator.nextId());
		protocol.opcode(requestControl.opcode());
		protocol.compressType(requestControl.compressType());
		protocol.serializeType(requestControl.serializeType());
		return protocol;
	}
	
	//less objects, don't need recycle
	public static RequestProtocol newHeartbeat() {
		RequestProtocol heartbeat = new RequestProtocol(null);
		heartbeat.packetId(PacketIdGenerator.nextId());
		heartbeat.heartbeat(true);
		return heartbeat;
	}
} public class ResponseProtocol implements Protocol{
	
	private static final Recycler<ResponseProtocol> RECYCLER = new Recycler<ResponseProtocol>(){
		@Override
		protected ResponseProtocol newObject(Recycler.Handle<ResponseProtocol> handle) {
			return new ResponseProtocol(handle);
		}
	};
	
    private static final int HEADER_LENGTH = 6;
    private static final int RESPONSE_FLAG = 0;
    private static final int HEARTBEAT_FLAG = 0x20;
        
    public static ResponseProtocol newInstance(){
    	return RECYCLER.get();
    }
    
    private final Recycler.Handle<ResponseProtocol> handle;
    
    //response direction
    private boolean heartbeat;
    private byte serializeType = SerializeType.NON_SERIALIZE;
    
    private int packetId;
    
    private byte result;
    private byte compressType = CompressType.NON_COMPRESS;

    private byte[] body;
    
    private SailfishException cause;
    
	public ResponseProtocol(Recycler.Handle<ResponseProtocol> handle) {
		this.handle = handle;
	}
    
	public void recycle(){
		if(null == handle){//some objects don't need recycle
			return;
		}
		
		heartbeat = false;
		serializeType = SerializeType.NON_SERIALIZE;
		packetId = 0;
		result = 0;
		compressType = CompressType.NON_COMPRESS;
		body = null;
		cause = null;
		handle.recycle(this);
	}
	
    @Override
    public void serialize(ByteBuf output) throws SailfishException {
        try{
            //write magic first
            output.writeShort(RemotingConstants.SAILFISH_MAGIC);
            //write package length(not contain current length field(4 bytes))
            if(this.heartbeat){
                output.writeInt(1);
            }else{
                output.writeInt(HEADER_LENGTH + bodyLength());
            }

            byte compactByte = (byte)RESPONSE_FLAG; 
            if(heartbeat){
                compactByte = (byte)(compactByte | HEARTBEAT_FLAG);
                output.writeByte(compactByte);
                return;
            }
            output.writeByte(compactByte | serializeType);
            
            output.writeInt(packetId);
            output.writeByte(result << 4 | compressType);
            
            if(bodyLength() != 0){
                output.writeBytes(body);
            }
        }catch(Throwable cause){
            throw new SailfishException(cause);
        }finally {
			recycle();
		}
    }

    @Override
    public void deserialize(ByteBuf input, int totalLength) throws SailfishException {
        try{
            byte compactByte = input.readByte();
            this.heartbeat = ((compactByte & HEARTBEAT_FLAG) != 0);
            if(this.heartbeat){
                return;
            }
            this.serializeType = (byte)(compactByte & 0x1F);
            
            this.packetId = input.readInt(); 
            
            byte tmp = input.readByte();

            this.result = (byte)(tmp >> 4 & 0xF);
            this.compressType = (byte)(tmp >> 0 & 0xF);
            
            //read body
            int bodyLength = totalLength - HEADER_LENGTH;
            if(bodyLength > 0){
                this.body = new byte[bodyLength];
                input.readBytes(this.body);
            }
        }catch(Throwable cause){
            throw new SailfishException(cause);
        }
    }

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

    public byte serializeType() {
        return serializeType;
    }

    public void serializeType(byte serializeType) {
        this.serializeType = ProtocolParameterChecker.checkSerializeType(serializeType);
    }

    public int packetId() {
        return packetId;
    }

    public void packetId(int packetId) {
        this.packetId = packetId;
    }

    public byte result() {
        return result;
    }

    public void result(byte result) {
        this.result = ProtocolParameterChecker.checkResult(result);
    }

    public byte compressType() {
        return compressType;
    }

    public void compressType(byte compressType) {
        this.compressType = ProtocolParameterChecker.checkCompressType(compressType);
    }

    public byte[] body() {
        return body;
    }

    public void body(byte[] body) {
        this.body = body;
    }

    public void errorStack(String errorStack) {
        if(StrUtils.isNotBlank(errorStack)){
            this.body = errorStack.getBytes(CharsetUtil.UTF_8);
        }
    }
   
    public void cause(SailfishException cause){
    	this.cause = cause;
    }
    
    public SailfishException cause(){
    	return this.cause;
    }
    
    private int bodyLength(){
        if(null == body){
            return 0;
        }
        return body.length;
    }

    @Override
    public boolean request() {
        return false;
    }
    
    public boolean heartbeat() {
        return this.heartbeat;
    }
    
	@Override
	public String toString() {
		return "ResponseProtocol [heartbeat=" + heartbeat + ", serializeType=" + serializeType + ", packetId="
				+ packetId + ", result=" + result + ", compressType=" + compressType + ", body=" + Arrays.toString(body)
				+ "]";
	}

	//less objects, don't need recycle
	public static ResponseProtocol newHeartbeat(){
        ResponseProtocol heartbeat = new ResponseProtocol(null);
        heartbeat.heartbeat(true);
        return heartbeat;
    }
    
	//less objects, don't need recycle
    public static ResponseProtocol newErrorResponse(int packetId, SailfishException cause){
        ResponseProtocol error = new ResponseProtocol(null);
        error.packetId(packetId);
        error.cause(cause);
        error.result(RemotingConstants.RESULT_FAIL);
        return error;
    }
} public class ChannelUtil {

	private static final Logger logger = LoggerFactory.getLogger(ChannelUtil.class);

	public static boolean clientSide(ChannelHandlerContext ctx) {
		Attribute<Boolean> clientSideAttr = ctx.channel().attr(ChannelAttrKeys.clientSide);
		return (null != clientSideAttr && null != clientSideAttr.get() && clientSideAttr.get());
	}

	public static void closeChannel(final Channel channel) {
		if (null == channel) {
			return;
		}
		channel.close().addListener(new ChannelFutureListener() {
			@Override
			public void operationComplete(ChannelFuture future) throws Exception {
				String log = String.format(
						"closeChannel: close connection to remoteAddress [%s] , localAddress [%s], ret:[%b]",
						null != channel.remoteAddress() ? channel.remoteAddress().toString() : "null",
						null != channel.localAddress() ? channel.localAddress().toString() : "null",
						future.isSuccess());
				logger.info(log);
			}
		});
	}
}
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.ReferenceCountUtil;
import io.netty.util.internal.PlatformDependent;
import sailfish.remoting.DefaultServer;
import sailfish.remoting.channel.ExchangeChannelGroup;
import sailfish.remoting.channel.ReadWriteServerExchangeChannelGroup;
import sailfish.remoting.channel.ReferenceCountedServerExchangeChannelGroup;
import sailfish.remoting.channel.ServerExchangeChannel;
import sailfish.remoting.channel.ServerExchangeChannelGroup;
import sailfish.remoting.configuration.NegotiateConfig;
import sailfish.remoting.constants.ChannelAttrKeys;
import sailfish.remoting.constants.ChannelAttrKeys.OneTime;
import sailfish.remoting.constants.Opcode;
import sailfish.remoting.protocol.Protocol;
import sailfish.remoting.protocol.RequestProtocol;
import sailfish.remoting.protocol.ResponseProtocol;
import sailfish.remoting.utils.ChannelUtil;
/**
 *
 *	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.handler;




/**
 * negotiate idleTimeout, maxIdleTimeout and settings about {@link NegotiateConfig} with remote peer
 * 
 * @author spccold
 * @version $Id: NegotiateChannelHandler.java, v 0.1 2016年11月23日 下午10:11:26 spccold Exp $
 */
@ChannelHandler.Sharable
public class NegotiateChannelHandler extends SimpleChannelInboundHandler<Protocol> {

	private static final Logger logger = LoggerFactory.getLogger(NegotiateChannelHandler.class);
	public static final NegotiateChannelHandler INSTANCE = new NegotiateChannelHandler();
	// This way we can reduce the memory usage compared to use Attributes.
	private final ConcurrentMap<ChannelHandlerContext, Boolean> negotiateMap = PlatformDependent.newConcurrentHashMap();

	public static final ConcurrentMap<String, ExchangeChannelGroup> uuid2ChannelGroup = new ConcurrentHashMap<>();

	private NegotiateChannelHandler() {
	};

	@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception {
		if (ChannelUtil.clientSide(ctx)) {
			negotiate(ctx);
		}
		ctx.fireChannelActive();
	}

	@Override
	public void channelInactive(ChannelHandlerContext ctx) throws Exception {
		if (!ChannelUtil.clientSide(ctx)) {
			String uuidStr = ctx.channel().attr(ChannelAttrKeys.uuidStr).get();