diff --git a/src/main/java/com/nhn/pinpoint/rpc/DefaultFuture.java b/src/main/java/com/nhn/pinpoint/rpc/DefaultFuture.java index 3364ce02f..31e60305c 100644 --- a/src/main/java/com/nhn/pinpoint/rpc/DefaultFuture.java +++ b/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 implements TimerTask, Future { @@ -22,8 +26,7 @@ public class DefaultFuture implements TimerTask, Future { private Timeout timeout; private FailureEventHandler failureEventHandler; - private FutureListener listener; - + private List> listeners = new ArrayList>(); public DefaultFuture() { this(3000); @@ -122,11 +125,13 @@ public class DefaultFuture implements TimerTask, Future { } private void notifyListener() { - FutureListener listener = this.listener; - if (listener != null) { - fireOnComplete(listener); - this.listener = null; - } + for (FutureListener listener : this.listeners) { + if (listener != null) { + fireOnComplete(listener); + } + } + + this.listeners = new ArrayList>(); } protected void notifyFailureHandle() { @@ -139,8 +144,8 @@ public class DefaultFuture implements TimerTask, Future { } @Override - public boolean setListener(FutureListener listener) { - if (listener == null) { + public boolean addListener(FutureListener listener) { + if (listener == null) { throw new NullPointerException("listener"); } @@ -149,13 +154,13 @@ public class DefaultFuture implements TimerTask, Future { if (ready) { alreadyReady = true; } else { - this.listener = listener; + this.listeners.add(listener); } } if (alreadyReady) { - fireOnComplete(listener); + fireOnComplete(listener); } return !alreadyReady; } diff --git a/src/main/java/com/nhn/pinpoint/rpc/Future.java b/src/main/java/com/nhn/pinpoint/rpc/Future.java index 22c2a7a74..21e54b7bf 100644 --- a/src/main/java/com/nhn/pinpoint/rpc/Future.java +++ b/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 { @@ -13,7 +14,7 @@ public interface Future { boolean isSuccess(); - boolean setListener(FutureListener listener); + boolean addListener(FutureListener listener); boolean await(long timeoutMillis); diff --git a/src/main/java/com/nhn/pinpoint/rpc/codec/PacketDecoder.java b/src/main/java/com/nhn/pinpoint/rpc/codec/PacketDecoder.java index c30aa8a26..235caeddb 100644 --- a/src/main/java/com/nhn/pinpoint/rpc/codec/PacketDecoder.java +++ b/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); + } } diff --git a/src/main/java/com/nhn/pinpoint/rpc/control/ControlMessageDecoder.java b/src/main/java/com/nhn/pinpoint/rpc/control/ControlMessageDecoder.java new file mode 100644 index 000000000..6f1a8fe25 --- /dev/null +++ b/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 answerList = new ArrayList(); + while (!isListFinished(in)) { + answerList.add(decode(in)); + } + in.get(); // Skip the terminator + return answerList; + case ControlMessageProtocolConstant.CONTROL_CHARACTER_MAP_START: + Map answerMap = new LinkedHashMap(); + 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; + } + +} diff --git a/src/main/java/com/nhn/pinpoint/rpc/control/ControlMessageEncoder.java b/src/main/java/com/nhn/pinpoint/rpc/control/ControlMessageEncoder.java new file mode 100644 index 000000000..251b13191 --- /dev/null +++ b/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 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 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 arrayToList = new ArrayList(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); + } + +} diff --git a/src/main/java/com/nhn/pinpoint/rpc/control/ControlMessageProtocolConstant.java b/src/main/java/com/nhn/pinpoint/rpc/control/ControlMessageProtocolConstant.java new file mode 100644 index 000000000..76f31b01f --- /dev/null +++ b/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'; + +} diff --git a/src/main/java/com/nhn/pinpoint/rpc/control/ProtocolException.java b/src/main/java/com/nhn/pinpoint/rpc/control/ProtocolException.java new file mode 100644 index 000000000..2d7abe514 --- /dev/null +++ b/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); + } + +} diff --git a/src/main/java/com/nhn/pinpoint/rpc/packet/ControlPacket.java b/src/main/java/com/nhn/pinpoint/rpc/packet/ControlPacket.java new file mode 100644 index 000000000..9c389ba54 --- /dev/null +++ b/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; + } + +} diff --git a/src/main/java/com/nhn/pinpoint/rpc/packet/ControlRegisterAgentConfirmPacket.java b/src/main/java/com/nhn/pinpoint/rpc/packet/ControlRegisterAgentConfirmPacket.java new file mode 100644 index 000000000..ee74c5104 --- /dev/null +++ b/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(); + } + +} diff --git a/src/main/java/com/nhn/pinpoint/rpc/packet/ControlRegisterAgentPacket.java b/src/main/java/com/nhn/pinpoint/rpc/packet/ControlRegisterAgentPacket.java new file mode 100644 index 000000000..4bfe41e2c --- /dev/null +++ b/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(); + } + +} diff --git a/src/main/java/com/nhn/pinpoint/rpc/packet/PacketType.java b/src/main/java/com/nhn/pinpoint/rpc/packet/PacketType.java index 95814ae31..7b6045df9 100644 --- a/src/main/java/com/nhn/pinpoint/rpc/packet/PacketType.java +++ b/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; } diff --git a/src/main/java/com/nhn/pinpoint/rpc/server/AgentProperties.java b/src/main/java/com/nhn/pinpoint/rpc/server/AgentProperties.java new file mode 100644 index 000000000..bc81bad88 --- /dev/null +++ b/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 properties; + + public AgentProperties(Map properties) { + this.properties = properties; + } + + public T getProperties(String key, Class returnClazz) { + Object value = properties.get(key); + + if (value == null) { + return null; + } + + if (ClassUtils.isAssignable(value.getClass(), returnClazz)) { + return (T) value; + } + + return null; + } + +} diff --git a/src/main/java/com/nhn/pinpoint/rpc/server/ChannelContext.java b/src/main/java/com/nhn/pinpoint/rpc/server/ChannelContext.java index c0fc82c21..cecf28cba 100644 --- a/src/main/java/com/nhn/pinpoint/rpc/server/ChannelContext.java +++ b/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; - } } diff --git a/src/main/java/com/nhn/pinpoint/rpc/server/PinpointServerSocket.java b/src/main/java/com/nhn/pinpoint/rpc/server/PinpointServerSocket.java index 7663ca176..8e911bbec 100644 --- a/src/main/java/com/nhn/pinpoint/rpc/server/PinpointServerSocket.java +++ b/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(); } diff --git a/src/main/java/com/nhn/pinpoint/rpc/server/PinpointServerSocketState.java b/src/main/java/com/nhn/pinpoint/rpc/server/PinpointServerSocketState.java new file mode 100644 index 000000000..12abde377 --- /dev/null +++ b/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; + } + +} diff --git a/src/main/java/com/nhn/pinpoint/rpc/server/PinpointServerSocketStateCode.java b/src/main/java/com/nhn/pinpoint/rpc/server/PinpointServerSocketStateCode.java new file mode 100644 index 000000000..44cded1f2 --- /dev/null +++ b/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 validBeforeStateSet; + + private PinpointServerSocketStateCode(PinpointServerSocketStateCode... validBeforeStates) { + this.validBeforeStateSet = new HashSet(); + + if (validBeforeStates != null) { + for (PinpointServerSocketStateCode eachStateCode : validBeforeStates) { + getValidBeforeStateSet().add(eachStateCode); + } + } + } + + public boolean canChangeState(PinpointServerSocketStateCode nextState) { + Set validBeforeStateSet = nextState.getValidBeforeStateSet(); + + if (validBeforeStateSet.contains(this)) { + return true; + } + + return false; + } + + public Set getValidBeforeStateSet() { + return validBeforeStateSet; + } + + public static boolean isRun(PinpointServerSocketStateCode code) { + if (code == RUN || code == RUN_WITHOUT_REGISTER) { + return true; + } + + return false; + } + +} diff --git a/src/main/java/com/nhn/pinpoint/rpc/util/ClassUtils.java b/src/main/java/com/nhn/pinpoint/rpc/util/ClassUtils.java new file mode 100644 index 000000000..0022171c2 --- /dev/null +++ b/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); + } + +} diff --git a/src/main/java/com/nhn/pinpoint/rpc/util/ControlMessageEnDeconderUtils.java b/src/main/java/com/nhn/pinpoint/rpc/util/ControlMessageEnDeconderUtils.java new file mode 100644 index 000000000..3cdfd2df1 --- /dev/null +++ b/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 value) throws ProtocolException { + return encoder.encode(value); + } + + public static Object decode(byte[] in) throws ProtocolException { + return decoder.decode(in); + } + +} diff --git a/src/test/java/com/nhn/pinpoint/rpc/FutureTest.java b/src/test/java/com/nhn/pinpoint/rpc/FutureTest.java index b16db8338..d1c6b6810 100644 --- a/src/test/java/com/nhn/pinpoint/rpc/FutureTest.java +++ b/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 future = new DefaultFuture(); + + SimpleListener listener1 = new SimpleListener(); + SimpleListener listener2 = new SimpleListener(); + + 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 future = new DefaultFuture(); + + SimpleListener listener = new SimpleListener(); + + future.setResult("Hello"); + + future.addListener(listener); + + Assert.assertTrue(listener.isFinished()); + } + + static class SimpleListener implements FutureListener { + + private final AtomicBoolean isFinished = new AtomicBoolean(false); + + @Override + public void onComplete(Future future) { + isFinished.compareAndSet(false, true); + } + + public boolean isFinished() { + return isFinished.get(); + } + } + } diff --git a/src/test/java/com/nhn/pinpoint/rpc/server/ControlPacketServerTest.java b/src/test/java/com/nhn/pinpoint/rpc/server/ControlPacketServerTest.java new file mode 100644 index 000000000..8da62c2ec --- /dev/null +++ b/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(); + } + +} diff --git a/src/test/java/com/nhn/pinpoint/rpc/server/PinpointServerSocketStateTest.java b/src/test/java/com/nhn/pinpoint/rpc/server/PinpointServerSocketStateTest.java new file mode 100644 index 000000000..082a4af7a --- /dev/null +++ b/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(); + } + +}