#4 Profiler - Collector간의 컨트롤 메시지 생성

1. 서버에 연결되는 채널의 상태 코드 추가
2. 에이전트 정보를 주고 받는 컨트롤 패킷 추가
3. Future Listener를 여러개 받을수 있게 수정
4. 컨트롤 메시지 인디코더 추가

src/main/java/com/nhn/pinpoint/rpc/DefaultFuture.java

@@ -1,5 +1,8 @@
 package com.nhn.pinpoint.rpc;
 
+import java.util.ArrayList;
+import java.util.List;
+
 import org.jboss.netty.util.Timeout;
 import org.jboss.netty.util.TimerTask;
 import org.slf4j.Logger;
@@ -7,6 +10,7 @@ import org.slf4j.LoggerFactory;
 
 /**
  * @author emeroad
+ * @author koo.taejin
  */
 public class DefaultFuture<T> implements TimerTask, Future<T> {
 
@@ -22,8 +26,7 @@ public class DefaultFuture<T> implements TimerTask, Future<T> {
 
     private Timeout timeout;
     private FailureEventHandler failureEventHandler;
-    private FutureListener<T> listener;
-
+    private List<FutureListener<T>> listeners = new ArrayList<FutureListener<T>>();
 
     public DefaultFuture() {
         this(3000);
@@ -122,11 +125,13 @@ public class DefaultFuture<T> implements TimerTask, Future<T> {
     }
 
     private void notifyListener() {
-        FutureListener<T> listener = this.listener;
-        if (listener != null) {
-            fireOnComplete(listener);
-            this.listener = null;
-        }
+    	for (FutureListener<T> listener : this.listeners) {
+            if (listener != null) {
+                fireOnComplete(listener);
+            }
+    	}
+    	
+    	this.listeners = new ArrayList<FutureListener<T>>();
     }
 
     protected void notifyFailureHandle() {
@@ -139,8 +144,8 @@ public class DefaultFuture<T> implements TimerTask, Future<T> {
     }
 
     @Override
-    public boolean setListener(FutureListener<T> listener) {
-        if (listener == null) {
+    public boolean addListener(FutureListener<T> listener) {
+    	if (listener == null) {
             throw new NullPointerException("listener");
         }
 
@@ -149,13 +154,13 @@ public class DefaultFuture<T> implements TimerTask, Future<T> {
             if (ready) {
                 alreadyReady = true;
             } else {
-                this.listener = listener;
+                this.listeners.add(listener);
             }
 
         }
 
         if (alreadyReady) {
-            fireOnComplete(listener);
+        	fireOnComplete(listener);
         }
         return !alreadyReady;
     }

src/main/java/com/nhn/pinpoint/rpc/Future.java

@@ -2,6 +2,7 @@ package com.nhn.pinpoint.rpc;
 
 /**
  * @author emeroad
+ * @author koo.taejin
  */
 public interface Future<T> {
 
@@ -13,7 +14,7 @@ public interface Future<T> {
 
     boolean isSuccess();
 
-    boolean setListener(FutureListener<T> listener);
+    boolean addListener(FutureListener<T> listener);
 
     boolean await(long timeoutMillis);
 

src/main/java/com/nhn/pinpoint/rpc/codec/PacketDecoder.java

@@ -2,6 +2,7 @@ package com.nhn.pinpoint.rpc.codec;
 
 import com.nhn.pinpoint.rpc.client.WriteFailFutureListener;
 import com.nhn.pinpoint.rpc.packet.*;
+
 import org.jboss.netty.buffer.ChannelBuffer;
 import org.jboss.netty.channel.Channel;
 import org.jboss.netty.channel.ChannelFuture;
@@ -12,6 +13,7 @@ import org.slf4j.LoggerFactory;
 
 /**
  * @author emeroad
+ * @author koo.taejin
  */
 public class PacketDecoder extends FrameDecoder {
 
@@ -56,13 +58,15 @@ public class PacketDecoder extends FrameDecoder {
                 readPong(packetType, buffer);
                 // pong 도 그냥 버리자.
                 return null;
+            case PacketType.CONTROL_REGISTER_AGENT:
+            	return readRegisterAgent(packetType, buffer);
         }
         logger.error("invalid packetType received. packetType:{}, channel:{}", packetType, channel);
         channel.close();
         return null;
     }
 
-    private void sendPong(Channel channel) {
+	private void sendPong(Channel channel) {
         // ping에 대한 응답으로 pong은 자동으로 응답한다.
         logger.debug("receive ping. send pong. {}", channel);
         ChannelFuture write = channel.write(PongPacket.PONG_PACKET);
@@ -123,6 +127,8 @@ public class PacketDecoder extends FrameDecoder {
         return StreamClosePacket.readBuffer(packetType, buffer);
     }
 
-
+    private Object readRegisterAgent(short packetType, ChannelBuffer buffer) {
+        return ControlRegisterAgentPacket.readBuffer(packetType, buffer);
+	}
 
 }

src/main/java/com/nhn/pinpoint/rpc/control/ControlMessageDecoder.java

@@ -0,0 +1,94 @@
+package com.nhn.pinpoint.rpc.control;
+
+import java.nio.ByteBuffer;
+import java.nio.charset.Charset;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * @author koo.taejin
+ */
+public class ControlMessageDecoder {
+
+	private Charset charset;
+
+	public ControlMessageDecoder() {
+		this.charset = Charset.forName("UTF-8");
+	}
+
+	public Object decode(byte[] in) throws ProtocolException {
+		return decode(ByteBuffer.wrap(in));
+	}
+		
+	public Object decode(ByteBuffer in) throws ProtocolException {
+		byte type = in.get();
+		switch (type) {
+		case ControlMessageProtocolConstant.TYPE_CHARACTER_NULL:
+			return null;
+		case ControlMessageProtocolConstant.TYPE_CHARACTER_BOOL_TRUE:
+			return Boolean.TRUE;
+		case ControlMessageProtocolConstant.TYPE_CHARACTER_BOOL_FALSE:
+			return Boolean.FALSE;
+		case ControlMessageProtocolConstant.TYPE_CHARACTER_INT:
+			return in.getInt();
+		case ControlMessageProtocolConstant.TYPE_CHARACTER_LONG:
+			return in.getLong();
+		case ControlMessageProtocolConstant.TYPE_CHARACTER_DOUBLE:
+			return Double.longBitsToDouble(in.getLong());
+		case ControlMessageProtocolConstant.TYPE_CHARACTER_STRING:
+			return decodeString(in);
+		case ControlMessageProtocolConstant.CONTROL_CHARACTER_LIST_START:
+			List<Object> answerList = new ArrayList<Object>();
+			while (!isListFinished(in)) {
+				answerList.add(decode(in));
+			}
+			in.get(); // Skip the terminator
+			return answerList;
+		case ControlMessageProtocolConstant.CONTROL_CHARACTER_MAP_START:
+			Map<Object, Object> answerMap = new LinkedHashMap<Object, Object>();
+			while (!isMapFinished(in)) {
+				Object key = decode(in);
+				Object value = decode(in);
+				answerMap.put(key, value);
+			}
+			in.get(); // Skip the terminator
+			return answerMap;
+		default:
+			throw new ProtocolException("invalid type character: " + (char) type + " (" + "0x" + Integer.toHexString(type) + ")");
+		}
+	}
+
+	private Object decodeString(ByteBuffer in) {
+		int length = readStringLength(in);
+
+		byte[] bytesToEncode = new byte[length];
+		in.get(bytesToEncode);
+
+		return new String(bytesToEncode, charset);
+	}
+
+	private boolean isMapFinished(ByteBuffer in) {
+		return in.get(in.position()) == ControlMessageProtocolConstant.CONTROL_CHARACTER_MAP_END;
+	}
+
+	private boolean isListFinished(ByteBuffer in) {
+		return in.get(in.position()) == ControlMessageProtocolConstant.CONTROL_CHARACTER_LIST_END;
+	}
+
+	private int readStringLength(ByteBuffer in) {
+		int result = 0;
+		int shift = 0;
+
+		while (true) {
+			byte b = in.get();
+			result |= (b & 0x7F) << shift;
+			if ((b & 0x80) != 128)
+				break;
+			shift += 7;
+		}
+		return result;
+	}
+
+}

src/main/java/com/nhn/pinpoint/rpc/control/ControlMessageEncoder.java

@@ -0,0 +1,179 @@
+package com.nhn.pinpoint.rpc.control;
+
+import java.lang.reflect.Array;
+import java.nio.charset.Charset;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+
+import org.jboss.netty.buffer.ChannelBuffer;
+import org.jboss.netty.buffer.ChannelBuffers;
+
+/**
+ * @author koo.taejin
+ */
+public class ControlMessageEncoder {
+
+	// 단순하게 가자 NPC랑 비슷 단) String에서 Chunk대신 bit 연산 사용하게 함
+	// UTF-8만 사용하게 함
+	private Charset charset;
+
+	public ControlMessageEncoder() {
+		this.charset = Charset.forName("UTF-8");
+	}
+
+	public byte[] encode(Map<String, Object> value) throws ProtocolException {
+		ChannelBuffer cb = ChannelBuffers.dynamicBuffer(100);
+		encode(value, cb);
+
+		int writeIndex = cb.writerIndex();
+		byte[] result = new byte[writeIndex];
+		
+		cb.readBytes(result);
+		
+		return result;
+	}
+	
+	private void encode(Map<String, Object> value, ChannelBuffer cb) throws ProtocolException {
+		encodeMap(value, cb);
+	}
+
+	private void encode(Object value, ChannelBuffer cb) throws ProtocolException {
+		try {
+			if (value == null) {
+				encodeNull(cb);
+			} else if (value instanceof String) {
+				encodeString((String) value, cb);
+			} else if (value instanceof Boolean) {
+				encodeBoolean((Boolean) value, cb);
+			} else if (value instanceof Short) {
+				encodeInt((Short) value, cb);
+			} else if (value instanceof Integer) {
+				encodeInt((Integer) value, cb);
+			} else if (value instanceof Long) {
+				encodeLong((Long) value, cb);
+			} else if (value instanceof Float) {
+				encodeDouble(((Float) value).doubleValue(), cb);
+			} else if (value instanceof Double) {
+				encodeDouble((Double) value, cb);
+			} else if (value instanceof Number) { // Other numbers (i.e.
+				// BigInteger and BigDecimal)
+				encodeString(value.toString(), cb);
+			} else if (value instanceof Collection) {
+				encodeCollection((Collection<?>) value, cb);
+			} else if (value instanceof Map) {
+				encodeMap((Map<?, ?>) value, cb);
+			} else if (value.getClass().isArray()) {
+				int arraySize = Array.getLength(value);
+
+				List<Object> arrayToList = new ArrayList<Object>(arraySize);
+				for (int i = 0; i < arraySize; i++) {
+					arrayToList.add(Array.get(value, i));
+				}
+				encodeCollection(arrayToList, cb);
+			} else {
+				throw new ProtocolException("Unsupported type : " + value.getClass().getName());
+			}
+		} catch (Exception e) {
+			throw new ProtocolException(e);
+		}
+	}
+
+	private void encodeNull(ChannelBuffer out) {
+		out.writeByte((byte) ControlMessageProtocolConstant.TYPE_CHARACTER_NULL);
+	}
+
+	private void encodeString(String value, ChannelBuffer out) {
+		out.writeByte((byte) ControlMessageProtocolConstant.TYPE_CHARACTER_STRING);
+		putPrefixedBytes(value.getBytes(charset), out);
+	}
+
+	private void encodeBoolean(boolean value, ChannelBuffer out) {
+		if (value) {
+			out.writeByte((byte) ControlMessageProtocolConstant.TYPE_CHARACTER_BOOL_TRUE);
+		} else {
+			out.writeByte((byte) ControlMessageProtocolConstant.TYPE_CHARACTER_BOOL_FALSE);
+		}
+	}
+
+	private void encodeInt(int value, ChannelBuffer out) {
+		out.writeByte((byte) ControlMessageProtocolConstant.TYPE_CHARACTER_INT);
+		
+		out.writeByte((byte) (value >> 24));
+		out.writeByte((byte) (value >> 16));
+		out.writeByte((byte) (value >> 8));
+		out.writeByte((byte) (value));
+	}
+
+	private void encodeLong(long value, ChannelBuffer out) {
+		out.writeByte((byte) ControlMessageProtocolConstant.TYPE_CHARACTER_LONG);
+		
+		out.writeByte((byte) (value >> 56));
+		out.writeByte((byte) (value >> 48));
+		out.writeByte((byte) (value >> 40));
+		out.writeByte((byte) (value >> 32));
+		out.writeByte((byte) (value >> 24));
+		out.writeByte((byte) (value >> 16));
+		out.writeByte((byte) (value >> 8));
+		out.writeByte((byte) (value));
+	}
+
+	private void encodeDouble(double value, ChannelBuffer out) {
+		out.writeByte((byte) ControlMessageProtocolConstant.TYPE_CHARACTER_DOUBLE);
+
+		long longValue = Double.doubleToLongBits(value);
+		
+		out.writeByte((byte) (longValue >> 56));
+		out.writeByte((byte) (longValue >> 48));
+		out.writeByte((byte) (longValue >> 40));
+		out.writeByte((byte) (longValue >> 32));
+		out.writeByte((byte) (longValue >> 24));
+		out.writeByte((byte) (longValue >> 16));
+		out.writeByte((byte) (longValue >> 8));
+		out.writeByte((byte) (longValue));
+	}
+
+	private void encodeCollection(Collection<?> collection, ChannelBuffer out) throws ProtocolException {
+		out.writeByte((byte) ControlMessageProtocolConstant.CONTROL_CHARACTER_LIST_START);
+		for (Object element : collection) {
+			encode(element, out);
+		}
+		out.writeByte((byte) ControlMessageProtocolConstant.CONTROL_CHARACTER_LIST_END);
+	}
+
+	private void encodeMap(Map<?, ?> map, ChannelBuffer out) throws ProtocolException {
+		out.writeByte((byte) ControlMessageProtocolConstant.CONTROL_CHARACTER_MAP_START);
+		for (Object element : map.entrySet()) {
+			Map.Entry<?, ?> entry = (Map.Entry<?, ?>) element;
+			encode(entry.getKey(), out);
+			encode(entry.getValue(), out);
+		}
+		out.writeByte((byte) ControlMessageProtocolConstant.CONTROL_CHARACTER_MAP_END);
+	}
+
+	private void putPrefixedBytes(byte[] value, ChannelBuffer out) {
+		int length = value.length;
+
+		byte[] lengthBuf = new byte[5];
+
+		int idx = 0;
+		while (true) {
+			if ((length & 0xFFFFFF80) == 0) {
+				lengthBuf[(idx++)] = (byte) length;
+				break;
+			}
+
+			lengthBuf[(idx++)] = (byte) (length & 0x7F | 0x80);
+
+			length >>>= 7;
+		}
+
+		for (int i = 0; i < idx; i++) {
+			out.writeByte(lengthBuf[i]);
+		}
+
+		out.writeBytes(value);
+	}
+
+}

src/main/java/com/nhn/pinpoint/rpc/control/ControlMessageProtocolConstant.java

@@ -0,0 +1,30 @@
+package com.nhn.pinpoint.rpc.control;
+
+/**
+ * @author koo.taejin
+ */
+public class ControlMessageProtocolConstant {
+
+	public static final int TYPE_CHARACTER_NULL = 'N';
+
+	public static final int TYPE_CHARACTER_BOOL_TRUE = 'T';
+
+	public static final int TYPE_CHARACTER_BOOL_FALSE = 'F';
+
+	public static final int TYPE_CHARACTER_INT = 'I';
+
+	public static final int TYPE_CHARACTER_LONG = 'L';
+
+	public static final int TYPE_CHARACTER_DOUBLE = 'D';
+
+	public static final int TYPE_CHARACTER_STRING = 'S';
+
+	public static final int CONTROL_CHARACTER_LIST_START = 'V';
+
+	public static final int CONTROL_CHARACTER_LIST_END = 'z';
+
+	public static final int CONTROL_CHARACTER_MAP_START = 'M';
+
+	public static final int CONTROL_CHARACTER_MAP_END = 'z';
+
+}

src/main/java/com/nhn/pinpoint/rpc/control/ProtocolException.java

@@ -0,0 +1,23 @@
+package com.nhn.pinpoint.rpc.control;
+
+/**
+ * @author koo.taejin
+ */
+public class ProtocolException extends Exception {
+
+    public ProtocolException() {
+    }
+
+    public ProtocolException(String message) {
+        super(message);
+    }
+
+    public ProtocolException(Throwable cause) {
+        super(cause);
+    }
+
+    public ProtocolException(String message, Throwable cause) {
+        super(message, cause);
+    }
+    
+}

src/main/java/com/nhn/pinpoint/rpc/packet/ControlPacket.java

@@ -0,0 +1,22 @@
+package com.nhn.pinpoint.rpc.packet;
+
+/**
+ * @author koo.taejin
+ */
+public abstract class ControlPacket extends BasicPacket {
+	
+	private int requestId;
+
+	public ControlPacket(byte[] payload) {
+		super(payload);
+	}
+
+	public int getRequestId() {
+		return requestId;
+	}
+
+	public void setRequestId(int requestId) {
+		this.requestId = requestId;
+	}
+
+}

src/main/java/com/nhn/pinpoint/rpc/packet/ControlRegisterAgentConfirmPacket.java

@@ -0,0 +1,68 @@
+package com.nhn.pinpoint.rpc.packet;
+
+import org.jboss.netty.buffer.ChannelBuffer;
+import org.jboss.netty.buffer.ChannelBuffers;
+
+/**
+ * @author koo.taejin
+ */
+public class ControlRegisterAgentConfirmPacket extends ControlPacket {
+
+	public ControlRegisterAgentConfirmPacket(byte[] payload) {
+		super(payload);
+	}
+
+	public ControlRegisterAgentConfirmPacket(int requestId, byte[] payload) {
+		super(payload);
+		setRequestId(requestId);
+	}
+
+	@Override
+	public short getPacketType() {
+		return PacketType.CONTROL_REGISTER_AGENT_CONFIRM;
+	}
+
+	@Override
+	public ChannelBuffer toBuffer() {
+
+		ChannelBuffer header = ChannelBuffers.buffer(2 + 4 + 4);
+		header.writeShort(PacketType.CONTROL_REGISTER_AGENT_CONFIRM);
+		header.writeInt(getRequestId());
+
+		return PayloadPacket.appendPayload(header, payload);
+	}
+
+	public static ControlRegisterAgentConfirmPacket readBuffer(short packetType, ChannelBuffer buffer) {
+		assert packetType == PacketType.CONTROL_REGISTER_AGENT_CONFIRM;
+
+		if (buffer.readableBytes() < 8) {
+			buffer.resetReaderIndex();
+			return null;
+		}
+
+		final int messageId = buffer.readInt();
+		final ChannelBuffer payload = PayloadPacket.readPayload(buffer);
+		if (payload == null) {
+			return null;
+		}
+		final ControlRegisterAgentConfirmPacket helloPacket = new ControlRegisterAgentConfirmPacket(payload.array());
+		helloPacket.setRequestId(messageId);
+		return helloPacket;
+	}
+
+	@Override
+	public String toString() {
+		final StringBuilder sb = new StringBuilder();
+		sb.append(this.getClass().getSimpleName());
+		sb.append("{requestId=").append(getRequestId());
+		sb.append(", ");
+		if (payload == null) {
+			sb.append("payload=null");
+		} else {
+			sb.append("payloadLength=").append(payload.length);
+		}
+		sb.append('}');
+		return sb.toString();
+	}
+
+}

src/main/java/com/nhn/pinpoint/rpc/packet/ControlRegisterAgentPacket.java

@@ -0,0 +1,68 @@
+package com.nhn.pinpoint.rpc.packet;
+
+import org.jboss.netty.buffer.ChannelBuffer;
+import org.jboss.netty.buffer.ChannelBuffers;
+
+/**
+ * @author koo.taejin
+ */
+public class ControlRegisterAgentPacket extends ControlPacket {
+
+	public ControlRegisterAgentPacket(byte[] payload) {
+		super(payload);
+	}
+
+	public ControlRegisterAgentPacket(int requestId, byte[] payload) {
+		super(payload);
+		setRequestId(requestId);
+	}
+
+	@Override
+	public short getPacketType() {
+		return PacketType.CONTROL_REGISTER_AGENT;
+	}
+
+	@Override
+	public ChannelBuffer toBuffer() {
+
+		ChannelBuffer header = ChannelBuffers.buffer(2 + 4 + 4);
+		header.writeShort(PacketType.CONTROL_REGISTER_AGENT);
+		header.writeInt(getRequestId());
+
+		return PayloadPacket.appendPayload(header, payload);
+	}
+
+	public static ControlRegisterAgentPacket readBuffer(short packetType, ChannelBuffer buffer) {
+		assert packetType == PacketType.CONTROL_REGISTER_AGENT;
+
+		if (buffer.readableBytes() < 8) {
+			buffer.resetReaderIndex();
+			return null;
+		}
+
+		final int messageId = buffer.readInt();
+		final ChannelBuffer payload = PayloadPacket.readPayload(buffer);
+		if (payload == null) {
+			return null;
+		}
+		final ControlRegisterAgentPacket helloPacket = new ControlRegisterAgentPacket(payload.array());
+		helloPacket.setRequestId(messageId);
+		return helloPacket;
+	}
+
+	@Override
+	public String toString() {
+		final StringBuilder sb = new StringBuilder();
+		sb.append(this.getClass().getSimpleName());
+		sb.append("{requestId=").append(getRequestId());
+		sb.append(", ");
+		if (payload == null) {
+			sb.append("payload=null");
+		} else {
+			sb.append("payloadLength=").append(payload.length);
+		}
+		sb.append('}');
+		return sb.toString();
+	}
+
+}

src/main/java/com/nhn/pinpoint/rpc/packet/PacketType.java

@@ -2,6 +2,7 @@ package com.nhn.pinpoint.rpc.packet;
 
 /**
  * @author emeroad
+ * @author koo.taejin
  */
 public class PacketType {
     public static final short APPLICATION_SEND = 1;
@@ -24,9 +25,15 @@ public class PacketType {
     public static final short CONTROL_CLIENT_CLOSE = 100;
     public static final short CONTROL_SERVER_CLOSE = 110;
 
+    // 컨트롤 패킷
+    public static final short CONTROL_REGISTER_AGENT = 150;
+    public static final short CONTROL_REGISTER_AGENT_CONFIRM = 151;
+
+    // ping, pong의 경우 성능상 두고 다른 CONTROL은 이걸로 뺌
     public static final short CONTROL_PING = 200;
     public static final short CONTROL_PONG = 201;
 
+    public static final short UNKNOWN = 500;
 
     public static final int PACKET_TYPE_SIZE = 2;
 }

src/main/java/com/nhn/pinpoint/rpc/server/AgentProperties.java

@@ -0,0 +1,42 @@
+package com.nhn.pinpoint.rpc.server;
+
+import java.util.Map;
+
+import com.nhn.pinpoint.rpc.util.ClassUtils;
+
+/**
+ * @author koo.taejin
+ */
+public class AgentProperties {
+
+	public static final String KEY_HOSTNAME = "hostName";
+	public static final String KEY_IP = "ip";
+	public static final String KEY_PORTS = "ports";
+	public static final String KEY_AGENTID = "agentId";
+	public static final String KEY_APPLICATION_NAME = "applicationName";
+	public static final String KEY_SERVICE_TYPE = "serviceType";
+	public static final String KEY_PID = "pid";
+	public static final String KEY_VERSION = "version";
+	public static final String KEY_START_TIME_MILLIS = "startTimestamp";
+
+	private final Map<String, Object> properties;
+
+	public AgentProperties(Map<String, Object> properties) {
+		this.properties = properties;
+	}
+
+	public <T> T getProperties(String key, Class<T> returnClazz) {
+		Object value = properties.get(key);
+
+		if (value == null) {
+			return null;
+		}
+
+		if (ClassUtils.isAssignable(value.getClass(), returnClazz)) {
+			return (T) value;
+		}
+
+		return null;
+	}
+
+}

src/main/java/com/nhn/pinpoint/rpc/server/ChannelContext.java

@@ -1,19 +1,26 @@
 package com.nhn.pinpoint.rpc.server;
 
 import org.jboss.netty.channel.Channel;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * @author emeroad
+ * @author koo.taejin
  */
 public class ChannelContext {
 
+	private final Logger logger = LoggerFactory.getLogger(this.getClass());
+	
     private final ServerStreamChannelManager streamChannelManager;
 
     private final Channel channel;
 
     private final SocketChannel socketChannel;
+    
+    private final PinpointServerSocketState state;
 
-    private volatile boolean closePacketReceived;
+    private AgentProperties agentProperties;
 
     public ChannelContext(Channel channel) {
         if (channel == null) {
@@ -22,6 +29,7 @@ public class ChannelContext {
         this.channel = channel;
         this.socketChannel = new SocketChannel(channel);
         this.streamChannelManager = new ServerStreamChannelManager(channel);
+        this.state = new PinpointServerSocketState();
     }
 
 
@@ -42,11 +50,26 @@ public class ChannelContext {
         return socketChannel;
     }
 
-    public boolean isClosePacketReceived() {
-        return closePacketReceived;
-    }
+	public PinpointServerSocketState getState() {
+		return state;
+	}
+	
+	public AgentProperties getAgentProperties() {
+		return agentProperties;
+	}
+
+	public boolean setAgentProperties(AgentProperties agentProperties) {
+		if (agentProperties == null) {
+			return false;
+		}
+		
+		if (this.agentProperties == null) {
+			this.agentProperties = agentProperties;
+			return true;
+		} 
+		
+		logger.warn("Already Register AgentProperties.({}).", this.agentProperties);
+		return false;
+	}
 
-    public void closePacketReceived() {
-        this.closePacketReceived = true;
-    }
 }

src/main/java/com/nhn/pinpoint/rpc/server/PinpointServerSocket.java

@@ -3,10 +3,13 @@ package com.nhn.pinpoint.rpc.server;
 import com.nhn.pinpoint.common.util.PinpointThreadFactory;
 import com.nhn.pinpoint.rpc.PinpointSocketException;
 import com.nhn.pinpoint.rpc.client.WriteFailFutureListener;
+import com.nhn.pinpoint.rpc.control.ProtocolException;
 import com.nhn.pinpoint.rpc.packet.*;
+import com.nhn.pinpoint.rpc.util.ControlMessageEnDeconderUtils;
 import com.nhn.pinpoint.rpc.util.CpuUtils;
 import com.nhn.pinpoint.rpc.util.LoggerFactorySetup;
 import com.nhn.pinpoint.rpc.util.TimerFactory;
+
 import org.jboss.netty.bootstrap.ServerBootstrap;
 import org.jboss.netty.channel.*;
 import org.jboss.netty.channel.group.ChannelGroup;
@@ -25,12 +28,15 @@ import org.slf4j.LoggerFactory;
 
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
+import java.util.HashMap;
+import java.util.Map;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 
 /**
  * @author emeroad
+ * @author koo.taejin
  */
 public class PinpointServerSocket extends SimpleChannelHandler {
 
@@ -105,6 +111,17 @@ public class PinpointServerSocket extends SimpleChannelHandler {
 
     }
 
+    private short getPacketType(Object packet) {
+    	if (packet == null) {
+    		return PacketType.UNKNOWN;
+    	}
+    	
+    	if (packet instanceof Packet) {
+    		return ((Packet) packet).getPacketType();
+    	}
+    	
+		return PacketType.UNKNOWN;
+    }
 
     private ServerBootstrap createBootStrap(int bossCount, int workerCount) {
         // profiler, collector,
@@ -120,18 +137,18 @@ public class PinpointServerSocket extends SimpleChannelHandler {
 
     @Override
     public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
-        Object message = e.getMessage();
-        if (message instanceof Packet) {
-            final Packet packet = (Packet) message;
-            final short packetType = packet.getPacketType();
-            final Channel channel = e.getChannel();
-            logger.debug("messageReceived:{} channel:{}", message, channel);
-            switch (packetType) {
-                case PacketType.APPLICATION_SEND: {
-                    SocketChannel socketChannel = getChannelContext(channel).getSocketChannel();
-                    messageListener.handleSend((SendPacket) message, socketChannel);
-                    return;
-                }
+    	final Channel channel = e.getChannel();
+    	Object message = e.getMessage();
+    	
+    	logger.debug("messageReceived:{} channel:{}", message, channel);
+    	
+		final short packetType = getPacketType(message);
+		switch (packetType) {
+			case PacketType.APPLICATION_SEND: {
+				SocketChannel socketChannel = getChannelContext(channel).getSocketChannel();
+				messageListener.handleSend((SendPacket) message, socketChannel);
+				return;
+			}
 //                case PacketType.APPLICATION_TRACE_SEND: {
 //                    SocketChannel socketChannel = getChannelContext(channel).getSocketChannel();
 //                    TraceSendPacket traceSendPacket = (TraceSendPacket) message;
@@ -144,35 +161,35 @@ public class PinpointServerSocket extends SimpleChannelHandler {
 //                    }
 //                    return;
 //                }
-                case PacketType.APPLICATION_REQUEST: {
-                    SocketChannel socketChannel = getChannelContext(channel).getSocketChannel();
-                    messageListener.handleRequest((RequestPacket) message, socketChannel);
-                    return;
-                }
-                case PacketType.APPLICATION_STREAM_CREATE:
-                case PacketType.APPLICATION_STREAM_CLOSE:
-                case PacketType.APPLICATION_STREAM_CREATE_SUCCESS:
-                case PacketType.APPLICATION_STREAM_CREATE_FAIL:
-                case PacketType.APPLICATION_STREAM_RESPONSE:
-                    handleStreamPacket((StreamPacket) message, channel);
-                    return;
-
-                case PacketType.CONTROL_CLIENT_CLOSE: {
-                    closeChannel(channel);
-                    return;
-                }
-                default:
-                    logger.warn("invalid messageReceived msg:{}, connection:{}", message, e.getChannel());
-            }
-        } else {
-            logger.warn("invalid messageReceived msg:{}, connection:{}", message, e.getChannel());
-        }
+			case PacketType.APPLICATION_REQUEST: {
+				SocketChannel socketChannel = getChannelContext(channel).getSocketChannel();
+				messageListener.handleRequest((RequestPacket) message, socketChannel);
+				return;
+			}
+			case PacketType.APPLICATION_STREAM_CREATE:
+			case PacketType.APPLICATION_STREAM_CLOSE:
+			case PacketType.APPLICATION_STREAM_CREATE_SUCCESS:
+			case PacketType.APPLICATION_STREAM_CREATE_FAIL:
+			case PacketType.APPLICATION_STREAM_RESPONSE:
+				handleStreamPacket((StreamPacket) message, channel);
+				return;
+			case PacketType.CONTROL_REGISTER_AGENT:
+				handleRegisterAgent((ControlRegisterAgentPacket) message, channel);
+				return;
+			case PacketType.CONTROL_CLIENT_CLOSE: {
+				closeChannel(channel);
+				return;
+			}
+			default:
+				logger.warn("invalid messageReceived msg:{}, connection:{}", message, e.getChannel());
+			}
     }
 
-    private void closeChannel(Channel channel) {
+	private void closeChannel(Channel channel) {
         logger.debug("received ClientClosePacket {}", channel);
         ChannelContext channelContext = getChannelContext(channel);
-        channelContext.closePacketReceived();
+        channelContext.getState().changeStateBeingShutdown();
+        
 //      상대방이 닫는거에 반응해서 socket을 닫도록 하자.
 //        channel.close();
     }
@@ -206,6 +223,35 @@ public class PinpointServerSocket extends SimpleChannelHandler {
         }
     }
 
+    private void handleRegisterAgent(ControlRegisterAgentPacket message, Channel channel) {
+        ChannelContext context = getChannelContext(channel);
+        byte[] payload = message.getPayload();
+        
+        try {
+			Map properties = (Map) ControlMessageEnDeconderUtils.decode(payload);
+			boolean isSuccess = context.setAgentProperties(new AgentProperties(properties));
+			// 이미 등록되어 있다면 상태를 변경하지 않음
+			if (isSuccess) {
+				context.getState().changeStateRun();
+			}
+			logger.debug("Channel({}) State changed to Run.", channel);
+		} catch (ProtocolException e) {
+			logger.warn(e.getMessage(), e);
+		}
+        
+		try {
+			Map result = new HashMap();
+			result.put("code", 0);
+
+			byte[] resultPayload = ControlMessageEnDeconderUtils.encode(result);
+			ControlRegisterAgentConfirmPacket packet = new ControlRegisterAgentConfirmPacket(message.getRequestId(), resultPayload);
+
+			channel.write(packet);
+		} catch (ProtocolException e) {
+			logger.warn(e.getMessage(), e);
+		}
+	}
+    
     @Override
     public void channelOpen(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
         final Channel channel = e.getChannel();
@@ -232,12 +278,25 @@ public class PinpointServerSocket extends SimpleChannelHandler {
             return;
         }
         prepareChannel(channel);
+        
+        ChannelContext channelContext = getChannelContext(channel);
+        channelContext.getState().changeStateRunWithoutRegister();
+        
         super.channelConnected(ctx, e);
     }
 
     @Override
     public void channelDisconnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
         final Channel channel = e.getChannel();
+        final ChannelContext channelContext = getChannelContext(channel);
+        PinpointServerSocketState state = channelContext.getState();
+        
+        if (state.getCurrentState() != PinpointServerSocketStateCode.BEING_SHUTDOWN) {
+        	state.changeStateShutdown();
+        } else {
+        	state.changeStateUnexpectedShutdown();
+        }
+        
         if (logger.isDebugEnabled()) {
             logger.debug("server channelDisconnected {}", channel);
         }
@@ -249,14 +308,19 @@ public class PinpointServerSocket extends SimpleChannelHandler {
     public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
         final Channel channel = e.getChannel();
         final ChannelContext channelContext = getChannelContext(channel);
-        if (channelContext.isClosePacketReceived()) {
+
+        PinpointServerSocketState state = channelContext.getState();
+        
+        if (state.getCurrentState() != PinpointServerSocketStateCode.BEING_SHUTDOWN) {
             if (logger.isDebugEnabled()) {
                 logger.debug("client channelClosed. normal closed. {}", channel);
             }
+        	state.changeStateShutdown();
         } else if(released) {
             if (logger.isDebugEnabled()) {
                 logger.debug("client channelClosed. server shutdown. {}", channel);
             }
+        	state.changeStateShutdown();
         } else {
             boolean check = checkIgnoreAddress(channel);
             if (check) {
@@ -264,6 +328,7 @@ public class PinpointServerSocket extends SimpleChannelHandler {
             } else {
                 logger.debug("checkAddress, Client channelClosed channelClosed {}", channel);
             }
+            state.changeStateUnexpectedShutdown();
         }
         channelContext.closeAllStreamChannel();
     }

src/main/java/com/nhn/pinpoint/rpc/server/PinpointServerSocketState.java

@@ -0,0 +1,69 @@
+package com.nhn.pinpoint.rpc.server;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * @author koo.taejin
+ */
+public class PinpointServerSocketState {
+
+	private final Logger logger = LoggerFactory.getLogger(this.getClass());
+
+	private PinpointServerSocketStateCode beforeState = PinpointServerSocketStateCode.NONE;
+	private PinpointServerSocketStateCode currentState = PinpointServerSocketStateCode.NONE;
+
+	private synchronized void setSessionState(PinpointServerSocketStateCode state) {
+		boolean enable = this.currentState.canChangeState(state);
+
+		if (enable) {
+			this.beforeState = this.currentState;
+			this.currentState = state;
+		} else {
+			PinpointServerSocketStateCode checkBefore = this.beforeState;
+			PinpointServerSocketStateCode checkCurrent = this.currentState;
+
+			String errorMessage = errorMessage(checkBefore, checkCurrent, state);
+			
+			this.beforeState = this.currentState;
+			this.currentState = PinpointServerSocketStateCode.ERROR_ILLEGAL_STATE_CHANGE;
+				
+			logger.warn(errorMessage);
+			
+			throw new IllegalStateException(errorMessage);
+		}
+	}
+	
+	public void changeStateRun() {
+		setSessionState(PinpointServerSocketStateCode.RUN);
+	}
+
+	public void changeStateRunWithoutRegister() {
+		setSessionState(PinpointServerSocketStateCode.RUN_WITHOUT_REGISTER);
+	}
+	
+	public void changeStateBeingShutdown() {
+		setSessionState(PinpointServerSocketStateCode.BEING_SHUTDOWN);
+	}
+
+	public void changeStateShutdown() {
+		setSessionState(PinpointServerSocketStateCode.SHUTDOWN);
+	}
+
+	public void changeStateUnexpectedShutdown() {
+		setSessionState(PinpointServerSocketStateCode.UNEXPECTED_SHUTDOWN);
+	}
+
+	public void changeStateUnkownError() {
+		setSessionState(PinpointServerSocketStateCode.ERROR_UNKOWN);
+	}
+	
+	private String errorMessage(PinpointServerSocketStateCode checkBefore, PinpointServerSocketStateCode checkCurrent, PinpointServerSocketStateCode nextState) {
+		return "Invalid State(current:" + checkCurrent + " before:" + checkBefore + " next:" + nextState;
+	}
+
+	public PinpointServerSocketStateCode getCurrentState() {
+		return currentState;
+	}
+
+}

src/main/java/com/nhn/pinpoint/rpc/server/PinpointServerSocketStateCode.java

@@ -0,0 +1,65 @@
+package com.nhn.pinpoint.rpc.server;
+
+import java.util.HashSet;
+import java.util.Set;
+
+/**
+ * @author koo.taejin
+ */
+public enum PinpointServerSocketStateCode {
+
+	// 상태는 다음과 같다.
+	// NONE : 아무 이벤트가 없는 상태
+	// RUN_WITHOUT_REGISTER_AGENT_RESOURCE : 동작은 하고 있지만 Agent의 정보를 서로 확인하지 못한 상태
+	// RUN : Agent의 정보를 서로 확인하고 동작 중인 상태
+	// BEING_SHUTDOWN : CLOSE 등의 명령을 받고 연결을 종료를 대기하는 상태 
+	// SHUTDOWN : 내가 끊거나 종료대기가 되어있는 상태일때 종료  
+	// UNEXPECTED_SHUTDOWN : CLOSE 등의 명령을 받지 못한 상태에서 상대방이 연결을 종료하였을떄 
+	
+	NONE(), 
+	RUN_WITHOUT_REGISTER(NONE), 
+	RUN(NONE, RUN_WITHOUT_REGISTER), 
+	BEING_SHUTDOWN(RUN, RUN_WITHOUT_REGISTER),
+	SHUTDOWN(RUN, RUN_WITHOUT_REGISTER, BEING_SHUTDOWN),
+	UNEXPECTED_SHUTDOWN(RUN, RUN_WITHOUT_REGISTER),
+	
+	// 서버쪽에서 먼저 연결을 끊자는 메시지도 필요하다. 
+	// 예를 들어 HELLO 이후 다 확인했는데, 같은 Agent명이 있으면(?) 이걸 사용자에게 말해야 할까? 아닐까? 알림 등
+	ERROR_UNKOWN(RUN, RUN_WITHOUT_REGISTER), 
+	ERROR_ILLEGAL_STATE_CHANGE(NONE, RUN, RUN_WITHOUT_REGISTER, BEING_SHUTDOWN, SHUTDOWN);
+
+	private final Set<PinpointServerSocketStateCode> validBeforeStateSet;
+
+	private PinpointServerSocketStateCode(PinpointServerSocketStateCode... validBeforeStates) {
+		this.validBeforeStateSet = new HashSet<PinpointServerSocketStateCode>();
+
+		if (validBeforeStates != null) {
+			for (PinpointServerSocketStateCode eachStateCode : validBeforeStates) {
+				getValidBeforeStateSet().add(eachStateCode);
+			}
+		}
+	}
+
+	public boolean canChangeState(PinpointServerSocketStateCode nextState) {
+		Set<PinpointServerSocketStateCode> validBeforeStateSet = nextState.getValidBeforeStateSet();
+
+		if (validBeforeStateSet.contains(this)) {
+			return true;
+		}
+
+		return false;
+	}
+
+	public Set<PinpointServerSocketStateCode> getValidBeforeStateSet() {
+		return validBeforeStateSet;
+	}
+
+	public static boolean isRun(PinpointServerSocketStateCode code) {
+		if (code == RUN || code == RUN_WITHOUT_REGISTER) {
+			return true;
+		}
+
+		return false;
+	}
+
+}

src/main/java/com/nhn/pinpoint/rpc/util/ClassUtils.java

@@ -0,0 +1,113 @@
+package com.nhn.pinpoint.rpc.util;
+
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Map;
+
+/**
+ * @author koo.taejin
+ */
+public class ClassUtils {
+
+	private static final Map primitiveWrapperMap = new HashMap();
+	static {
+		primitiveWrapperMap.put(Boolean.TYPE, Boolean.class);
+		primitiveWrapperMap.put(Byte.TYPE, Byte.class);
+		primitiveWrapperMap.put(Character.TYPE, Character.class);
+		primitiveWrapperMap.put(Short.TYPE, Short.class);
+		primitiveWrapperMap.put(Integer.TYPE, Integer.class);
+		primitiveWrapperMap.put(Long.TYPE, Long.class);
+		primitiveWrapperMap.put(Double.TYPE, Double.class);
+		primitiveWrapperMap.put(Float.TYPE, Float.class);
+		primitiveWrapperMap.put(Void.TYPE, Void.TYPE);
+	}
+	
+    private static final Map wrapperPrimitiveMap = new HashMap();
+    static {
+        for (Iterator it = primitiveWrapperMap.keySet().iterator(); it.hasNext();) {
+            Class primitiveClass = (Class) it.next();
+            Class wrapperClass = (Class) primitiveWrapperMap.get(primitiveClass);
+            if (!primitiveClass.equals(wrapperClass)) {
+                wrapperPrimitiveMap.put(wrapperClass, primitiveClass);
+            }
+        }
+    }
+
+	public static boolean isAssignable(Class cls, Class toClass) {
+		return isAssignable(cls, toClass, true);
+	}
+		
+	public static boolean isAssignable(Class cls, Class toClass, boolean autoboxing) {
+		if (toClass == null) {
+			return false;
+		}
+		// have to check for null, as isAssignableFrom doesn't
+		if (cls == null) {
+			return !(toClass.isPrimitive());
+		}
+		// autoboxing:
+		if (autoboxing) {
+			if (cls.isPrimitive() && !toClass.isPrimitive()) {
+				cls = primitiveToWrapper(cls);
+				if (cls == null) {
+					return false;
+				}
+			}
+			if (toClass.isPrimitive() && !cls.isPrimitive()) {
+				cls = wrapperToPrimitive(cls);
+				if (cls == null) {
+					return false;
+				}
+			}
+		}
+		if (cls.equals(toClass)) {
+			return true;
+		}
+		if (cls.isPrimitive()) {
+			if (toClass.isPrimitive() == false) {
+				return false;
+			}
+			if (Integer.TYPE.equals(cls)) {
+				return Long.TYPE.equals(toClass) || Float.TYPE.equals(toClass) || Double.TYPE.equals(toClass);
+			}
+			if (Long.TYPE.equals(cls)) {
+				return Float.TYPE.equals(toClass) || Double.TYPE.equals(toClass);
+			}
+			if (Boolean.TYPE.equals(cls)) {
+				return false;
+			}
+			if (Double.TYPE.equals(cls)) {
+				return false;
+			}
+			if (Float.TYPE.equals(cls)) {
+				return Double.TYPE.equals(toClass);
+			}
+			if (Character.TYPE.equals(cls)) {
+				return Integer.TYPE.equals(toClass) || Long.TYPE.equals(toClass) || Float.TYPE.equals(toClass) || Double.TYPE.equals(toClass);
+			}
+			if (Short.TYPE.equals(cls)) {
+				return Integer.TYPE.equals(toClass) || Long.TYPE.equals(toClass) || Float.TYPE.equals(toClass) || Double.TYPE.equals(toClass);
+			}
+			if (Byte.TYPE.equals(cls)) {
+				return Short.TYPE.equals(toClass) || Integer.TYPE.equals(toClass) || Long.TYPE.equals(toClass) || Float.TYPE.equals(toClass)
+						|| Double.TYPE.equals(toClass);
+			}
+			// should never get here
+			return false;
+		}
+		return toClass.isAssignableFrom(cls);
+	}
+
+	public static Class primitiveToWrapper(Class cls) {
+		Class convertedClass = cls;
+		if (cls != null && cls.isPrimitive()) {
+			convertedClass = (Class) primitiveWrapperMap.get(cls);
+		}
+		return convertedClass;
+	}
+
+	public static Class wrapperToPrimitive(Class cls) {
+		return (Class) wrapperPrimitiveMap.get(cls);
+	}
+
+}

src/main/java/com/nhn/pinpoint/rpc/util/ControlMessageEnDeconderUtils.java

@@ -0,0 +1,28 @@
+package com.nhn.pinpoint.rpc.util;
+
+import java.util.Map;
+
+import com.nhn.pinpoint.rpc.control.ControlMessageDecoder;
+import com.nhn.pinpoint.rpc.control.ControlMessageEncoder;
+import com.nhn.pinpoint.rpc.control.ProtocolException;
+
+/**
+ * @author koo.taejin
+ */
+public class ControlMessageEnDeconderUtils {
+
+	private static final ControlMessageEncoder encoder = new ControlMessageEncoder();
+	private static final ControlMessageDecoder decoder = new ControlMessageDecoder();
+
+	private ControlMessageEnDeconderUtils() {
+	}
+
+	public static byte[] encode(Map<String, Object> value) throws ProtocolException {
+		return encoder.encode(value);
+	}
+
+	public static Object decode(byte[] in) throws ProtocolException {
+		return decoder.decode(in);
+	}
+
+}

src/test/java/com/nhn/pinpoint/rpc/FutureTest.java

@@ -1,7 +1,60 @@
 package com.nhn.pinpoint.rpc;
 
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.junit.Assert;
+import org.junit.Test;
+
 /**
  * @author emeroad
+ * @author koo.taejin
  */
 public class FutureTest {
+
+	@Test
+	public void simpleTest1() {
+		DefaultFuture<String> future = new DefaultFuture<String>();
+
+		SimpleListener<String> listener1 = new SimpleListener<String>();
+		SimpleListener<String> listener2 = new SimpleListener<String>();
+
+		future.addListener(listener1);
+		future.addListener(listener2);
+
+		Assert.assertFalse(listener1.isFinished());
+		Assert.assertFalse(listener2.isFinished());
+
+		future.setResult("Hello");
+
+		Assert.assertTrue(listener1.isFinished());
+		Assert.assertTrue(listener2.isFinished());
+	}
+
+	@Test
+	public void simpleTest2() {
+		DefaultFuture<String> future = new DefaultFuture<String>();
+
+		SimpleListener<String> listener = new SimpleListener<String>();
+
+		future.setResult("Hello");
+
+		future.addListener(listener);
+
+		Assert.assertTrue(listener.isFinished());
+	}
+
+	static class SimpleListener<T> implements FutureListener<T> {
+
+		private final AtomicBoolean isFinished = new AtomicBoolean(false);
+
+		@Override
+		public void onComplete(Future<T> future) {
+			isFinished.compareAndSet(false, true);
+		}
+
+		public boolean isFinished() {
+			return isFinished.get();
+		}
+	}
+
 }

src/test/java/com/nhn/pinpoint/rpc/server/ControlPacketServerTest.java

@@ -0,0 +1,58 @@
+package com.nhn.pinpoint.rpc.server;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.net.Socket;
+import java.nio.ByteBuffer;
+import java.util.Collections;
+import java.util.Map;
+
+import org.jboss.netty.buffer.ChannelBuffer;
+import org.jboss.netty.buffer.ChannelBuffers;
+import org.junit.Assert;
+import org.junit.Test;
+
+import com.nhn.pinpoint.rpc.packet.ControlRegisterAgentConfirmPacket;
+import com.nhn.pinpoint.rpc.packet.ControlRegisterAgentPacket;
+import com.nhn.pinpoint.rpc.util.ControlMessageEnDeconderUtils;
+
+/**
+ * @author koo.taejin
+ */
+public class ControlPacketServerTest {
+
+	@Test
+	public void registerAgentTest() throws Exception {
+		PinpointServerSocket pinpointServerSocket = new PinpointServerSocket();
+		pinpointServerSocket.bind("127.0.0.1", 22234);
+
+		Socket socket = new Socket("127.0.0.1", 22234);
+
+		byte[] payload = ControlMessageEnDeconderUtils.encode(Collections.EMPTY_MAP);
+		ControlRegisterAgentPacket packet = new ControlRegisterAgentPacket(1, payload);
+
+		ByteBuffer bb = packet.toBuffer().toByteBuffer(0, packet.toBuffer().writerIndex());
+
+		sendData(socket.getOutputStream(), bb.array());
+
+		byte[] a = new byte[24];
+		socket.getInputStream().read(a);
+
+		ChannelBuffer cb = ChannelBuffers.wrappedBuffer(a);
+		short packetType = cb.readShort();
+
+		ControlRegisterAgentConfirmPacket p = ControlRegisterAgentConfirmPacket.readBuffer(packetType, cb);
+		Map result = (Map) ControlMessageEnDeconderUtils.decode(p.getPayload());
+
+		Assert.assertEquals(0, result.get("code"));
+
+		socket.close();
+		pinpointServerSocket.close();
+	}
+
+	private void sendData(OutputStream os, byte[] payload) throws IOException {
+		os.write(payload);
+		os.flush();
+	}
+
+}

src/test/java/com/nhn/pinpoint/rpc/server/PinpointServerSocketStateTest.java

@@ -0,0 +1,105 @@
+package com.nhn.pinpoint.rpc.server;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+/**
+ * @author koo.taejin
+ */
+public class PinpointServerSocketStateTest {
+
+	// Agent버전이 최신일 경우 (2014-07 기준)
+	// 가장 기본적인 형태 RUN_WITHOUT_RGISTER -> 이후 Agent 정보 획득 RUN ->  클라이언트 종료전 Agent 정보 제거 요청에 따른 BEING_SHUTDOWN -> 연결종료
+	@Test
+	public void changeStateTest1() {
+		PinpointServerSocketState state = new PinpointServerSocketState();
+
+		state.changeStateRunWithoutRegister();
+		Assert.assertEquals(PinpointServerSocketStateCode.RUN_WITHOUT_REGISTER, state.getCurrentState());
+
+		state.changeStateRun();
+		Assert.assertEquals(PinpointServerSocketStateCode.RUN, state.getCurrentState());
+
+		state.changeStateBeingShutdown();
+		Assert.assertEquals(PinpointServerSocketStateCode.BEING_SHUTDOWN, state.getCurrentState());
+
+		state.changeStateShutdown();
+		Assert.assertEquals(PinpointServerSocketStateCode.SHUTDOWN, state.getCurrentState());
+	}
+
+	// Agent버전이 최신일 경우 (2014-07 기준)
+	// 가장 기본적인 형태 RUN -> 이후 Agent 정보 획득 RUN ->  클라이언트 종료전 Agent 정보 제거 요청에 따른 BEING_SHUTDOWN -> 연결종료
+	@Test
+	public void changeStateTest2() {
+		PinpointServerSocketState state = new PinpointServerSocketState();
+
+		state.changeStateRun();
+		Assert.assertEquals(PinpointServerSocketStateCode.RUN, state.getCurrentState());
+
+		state.changeStateBeingShutdown();
+		Assert.assertEquals(PinpointServerSocketStateCode.BEING_SHUTDOWN, state.getCurrentState());
+
+		state.changeStateShutdown();
+		Assert.assertEquals(PinpointServerSocketStateCode.SHUTDOWN, state.getCurrentState());
+	}
+
+	// Agent버전이 구버전일 경우 (2014-07 기준)
+	// 가장 기본적인 형태 RUN_WITHOUT_RGISTER -> 연결종료
+	@Test
+	public void changeStateTest3() {
+		PinpointServerSocketState state = new PinpointServerSocketState();
+
+		state.changeStateRunWithoutRegister();
+		Assert.assertEquals(PinpointServerSocketStateCode.RUN_WITHOUT_REGISTER, state.getCurrentState());
+
+		state.changeStateUnexpectedShutdown();
+		Assert.assertEquals(PinpointServerSocketStateCode.UNEXPECTED_SHUTDOWN, state.getCurrentState());
+	}
+
+	// Agent버전이 구버전일 경우 (2014-07 기준)
+	// 가장 기본적인 형태 RUN_WITHOUT_RGISTER -> 연결종료
+	@Test
+	public void changeStateTest4() {
+		PinpointServerSocketState state = new PinpointServerSocketState();
+
+		state.changeStateRunWithoutRegister();
+		Assert.assertEquals(PinpointServerSocketStateCode.RUN_WITHOUT_REGISTER, state.getCurrentState());
+
+		state.changeStateShutdown();
+		Assert.assertEquals(PinpointServerSocketStateCode.SHUTDOWN, state.getCurrentState());
+	}
+
+	@Test
+	public void changeStateTest5() {
+		PinpointServerSocketState state = new PinpointServerSocketState();
+
+		state.changeStateRun();
+		Assert.assertEquals(PinpointServerSocketStateCode.RUN, state.getCurrentState());
+
+		state.changeStateShutdown();
+		Assert.assertEquals(PinpointServerSocketStateCode.SHUTDOWN, state.getCurrentState());
+	}
+	
+	
+
+	@Test(expected = IllegalStateException.class)
+	public void invalidChangeStateTest1() {
+		PinpointServerSocketState state = new PinpointServerSocketState();
+
+		state.changeStateBeingShutdown();
+	}
+
+	@Test(expected = IllegalStateException.class)
+	public void invalidChangeStateTest2() {
+		PinpointServerSocketState state = new PinpointServerSocketState();
+
+		state.changeStateRun();
+		Assert.assertEquals(PinpointServerSocketStateCode.RUN, state.getCurrentState());
+
+		state.changeStateBeingShutdown();
+		Assert.assertEquals(PinpointServerSocketStateCode.BEING_SHUTDOWN, state.getCurrentState());
+
+		state.changeStateUnexpectedShutdown();
+	}
+
+}