From df238070323f34ef4c8281e17dff014382a3aedd Mon Sep 17 00:00:00 2001 From: Woonduk Kang Date: Fri, 5 Jul 2013 10:52:27 +0000 Subject: [PATCH] =?UTF-8?q?[=EA=B0=95=EC=9A=B4=EB=8D=95]=20[LUCYSUS-1744]?= =?UTF-8?q?=20stream=20=EC=B1=84=EB=84=90=EA=B4=80=EB=A6=AC=20=EA=B8=B0?= =?UTF-8?q?=EB=8A=A5=20=EA=B0=9C=EB=B0=9C?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit git-svn-id: http://svn.bds.nhncorp.com/pe/hippo-commons/trunk@1952 84d0f5b1-2673-498c-a247-62c4ff18d310 --- ...{MessageFuture.java => DefaultFuture.java} | 55 ++++++------ .../common/io/rpc/FailureEventHandler.java | 9 ++ .../pinpoint/common/io/rpc/FailureHandle.java | 9 -- .../nhn/pinpoint/common/io/rpc/Future.java | 20 +++++ .../common/io/rpc/FutureListener.java | 8 ++ .../common/io/rpc/MessageFutureListener.java | 11 --- .../common/io/rpc/PinpointServerSocket.java | 7 +- .../common/io/rpc/PinpointSocket.java | 58 +++++++++---- .../common/io/rpc/PinpointSocketFactory.java | 9 +- .../common/io/rpc/RequestProcessor.java | 43 +++++++--- .../common/io/rpc/ServerMessageListener.java | 9 +- .../io/rpc/SimpleSeverMessageListener.java | 15 +--- .../common/io/rpc/SocketRequestHandler.java | 53 ++++-------- .../pinpoint/common/io/rpc/StreamChannel.java | 85 +++++++++++++++---- .../common/io/rpc/StreamChannelFuture.java | 16 ---- ...patcher.java => StreamChannelManager.java} | 34 ++++++-- ...MessageFutureTest.java => FutureTest.java} | 2 +- .../io/rpc/PinpointSocketFactoryTest.java | 9 +- .../common/io/rpc/RequestProcessorTest.java | 16 ++-- 19 files changed, 280 insertions(+), 188 deletions(-) rename src/main/java/com/nhn/pinpoint/common/io/rpc/{MessageFuture.java => DefaultFuture.java} (73%) create mode 100644 src/main/java/com/nhn/pinpoint/common/io/rpc/FailureEventHandler.java delete mode 100644 src/main/java/com/nhn/pinpoint/common/io/rpc/FailureHandle.java create mode 100644 src/main/java/com/nhn/pinpoint/common/io/rpc/Future.java create mode 100644 src/main/java/com/nhn/pinpoint/common/io/rpc/FutureListener.java delete mode 100644 src/main/java/com/nhn/pinpoint/common/io/rpc/MessageFutureListener.java delete mode 100644 src/main/java/com/nhn/pinpoint/common/io/rpc/StreamChannelFuture.java rename src/main/java/com/nhn/pinpoint/common/io/rpc/{StreamPacketDispatcher.java => StreamChannelManager.java} (57%) rename src/test/java/com/nhn/pinpoint/common/io/rpc/{MessageFutureTest.java => FutureTest.java} (57%) diff --git a/src/main/java/com/nhn/pinpoint/common/io/rpc/MessageFuture.java b/src/main/java/com/nhn/pinpoint/common/io/rpc/DefaultFuture.java similarity index 73% rename from src/main/java/com/nhn/pinpoint/common/io/rpc/MessageFuture.java rename to src/main/java/com/nhn/pinpoint/common/io/rpc/DefaultFuture.java index c9e5e0e61..6d081acaf 100644 --- a/src/main/java/com/nhn/pinpoint/common/io/rpc/MessageFuture.java +++ b/src/main/java/com/nhn/pinpoint/common/io/rpc/DefaultFuture.java @@ -8,60 +8,62 @@ import org.slf4j.LoggerFactory; /** * */ -public class MessageFuture implements TimerTask { +public class DefaultFuture implements TimerTask, Future { - private static final Logger logger = LoggerFactory.getLogger(MessageFuture.class); + private static final Logger logger = LoggerFactory.getLogger(DefaultFuture.class); private long timeoutMillis; private int waiters = 0; private boolean ready = false; - private int requestId; - private Message message; + private T object; private Throwable cause; private Timeout timeout; - private FailureHandle failureHandle; - private MessageFutureListener listener; + private FailureEventHandler failureEventHandler; + private FutureListener listener; - public MessageFuture(int requestId) { - this.requestId = requestId; + public DefaultFuture() { + this(3000); } - public MessageFuture(int requestId, long timeoutMillis) { - this.requestId = requestId; + public DefaultFuture(long timeoutMillis) { this.timeoutMillis = timeoutMillis; } - public synchronized Message getMessage() { + @Override + public synchronized T getObject() { if (this.cause != null) { throw new PinpointSocketException(cause); } - return message; + return object; } + @Override public synchronized Throwable getCause() { return cause; } + @Override public synchronized boolean isReady() { return ready; } + @Override public synchronized boolean isSuccess() { return ready && cause == null; } - public boolean setMessage(Message message) { + public boolean setObject(T message) { synchronized (this) { if (ready) { return false; } this.ready = true; - this.message = message; + this.object = message; if (waiters > 0) { notifyAll(); } @@ -120,23 +122,24 @@ public class MessageFuture implements TimerTask { } private void notifyListener() { - MessageFutureListener listener = this.listener; + FutureListener listener = this.listener; if (listener != null) { fireOnComplete(listener); this.listener = null; } } - private void notifyFailureHandle() { + protected void notifyFailureHandle() { - FailureHandle failureHandle = this.failureHandle; - if (failureHandle != null) { - failureHandle.handleFailure(this.requestId); - this.failureHandle = null; + FailureEventHandler failureEventHandler = this.failureEventHandler; + if (failureEventHandler != null) { + failureEventHandler.fireFailure(); + this.failureEventHandler = null; } } - public boolean setListener(MessageFutureListener listener) { + @Override + public boolean setListener(FutureListener listener) { if (listener == null) { throw new NullPointerException("listener"); } @@ -157,20 +160,22 @@ public class MessageFuture implements TimerTask { return !alreadyReady; } - private boolean fireOnComplete(MessageFutureListener listener) { + private boolean fireOnComplete(FutureListener listener) { try { listener.onComplete(this); return true; } catch (Throwable th) { - logger.warn("MessageFutureListener.onComplete() fail Caused:{}", th.getMessage(), th); + logger.warn("FutureListener.onComplete() fail Caused:{}", th.getMessage(), th); return false; } } + @Override public boolean await(long timeoutMillis) { return await0(timeoutMillis); } + @Override public boolean await() { return await0(this.timeoutMillis); } @@ -208,7 +213,7 @@ public class MessageFuture implements TimerTask { this.timeout = timeout; } - public void setFailureHandle(FailureHandle failureHandle) { - this.failureHandle = failureHandle; + public void setFailureEventHandler(FailureEventHandler failureEventHandler) { + this.failureEventHandler = failureEventHandler; } } diff --git a/src/main/java/com/nhn/pinpoint/common/io/rpc/FailureEventHandler.java b/src/main/java/com/nhn/pinpoint/common/io/rpc/FailureEventHandler.java new file mode 100644 index 000000000..b3a2de59b --- /dev/null +++ b/src/main/java/com/nhn/pinpoint/common/io/rpc/FailureEventHandler.java @@ -0,0 +1,9 @@ +package com.nhn.pinpoint.common.io.rpc; + +/** + * + */ +public interface FailureEventHandler { + + boolean fireFailure(); +} diff --git a/src/main/java/com/nhn/pinpoint/common/io/rpc/FailureHandle.java b/src/main/java/com/nhn/pinpoint/common/io/rpc/FailureHandle.java deleted file mode 100644 index 3452e75e0..000000000 --- a/src/main/java/com/nhn/pinpoint/common/io/rpc/FailureHandle.java +++ /dev/null @@ -1,9 +0,0 @@ -package com.nhn.pinpoint.common.io.rpc; - -/** - * - */ -public interface FailureHandle { - - void handleFailure(int requestId); -} diff --git a/src/main/java/com/nhn/pinpoint/common/io/rpc/Future.java b/src/main/java/com/nhn/pinpoint/common/io/rpc/Future.java new file mode 100644 index 000000000..ca67aa087 --- /dev/null +++ b/src/main/java/com/nhn/pinpoint/common/io/rpc/Future.java @@ -0,0 +1,20 @@ +package com.nhn.pinpoint.common.io.rpc; + +/** + * + */ +public interface Future { + T getObject(); + + Throwable getCause(); + + boolean isReady(); + + boolean isSuccess(); + + boolean setListener(FutureListener listener); + + boolean await(long timeoutMillis); + + boolean await(); +} diff --git a/src/main/java/com/nhn/pinpoint/common/io/rpc/FutureListener.java b/src/main/java/com/nhn/pinpoint/common/io/rpc/FutureListener.java new file mode 100644 index 000000000..9bba1e437 --- /dev/null +++ b/src/main/java/com/nhn/pinpoint/common/io/rpc/FutureListener.java @@ -0,0 +1,8 @@ +package com.nhn.pinpoint.common.io.rpc; + +/** + * + */ +public interface FutureListener { + void onComplete(Future future); +} diff --git a/src/main/java/com/nhn/pinpoint/common/io/rpc/MessageFutureListener.java b/src/main/java/com/nhn/pinpoint/common/io/rpc/MessageFutureListener.java deleted file mode 100644 index 9c55a2540..000000000 --- a/src/main/java/com/nhn/pinpoint/common/io/rpc/MessageFutureListener.java +++ /dev/null @@ -1,11 +0,0 @@ -package com.nhn.pinpoint.common.io.rpc; - -import org.jboss.netty.channel.ChannelHandlerContext; -import org.jboss.netty.channel.MessageEvent; - -/** - * - */ -public interface MessageFutureListener { - void onComplete(MessageFuture messageFuture); -} diff --git a/src/main/java/com/nhn/pinpoint/common/io/rpc/PinpointServerSocket.java b/src/main/java/com/nhn/pinpoint/common/io/rpc/PinpointServerSocket.java index 93f4d46be..7f7f0e503 100644 --- a/src/main/java/com/nhn/pinpoint/common/io/rpc/PinpointServerSocket.java +++ b/src/main/java/com/nhn/pinpoint/common/io/rpc/PinpointServerSocket.java @@ -2,6 +2,8 @@ package com.nhn.pinpoint.common.io.rpc; import com.nhn.pinpoint.common.io.rpc.packet.RequestPacket; import com.nhn.pinpoint.common.io.rpc.packet.SendPacket; +import com.nhn.pinpoint.common.io.rpc.packet.StreamCreatePacket; +import com.nhn.pinpoint.common.io.rpc.packet.StreamPacket; import org.jboss.netty.bootstrap.ServerBootstrap; import org.jboss.netty.channel.*; import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory; @@ -75,7 +77,10 @@ public class PinpointServerSocket extends SimpleChannelHandler { listener.handleSend((SendPacket) message, e.getChannel()); } else if (message instanceof RequestPacket) { listener.handleRequest((RequestPacket) message, e.getChannel()); - } else { + } else if(message instanceof StreamPacket) { + listener.handleStream((StreamPacket) message, e.getChannel()); + } + else { logger.error("invalid messageReceived msg:{}, connection:{}", message, e.getChannel()); } diff --git a/src/main/java/com/nhn/pinpoint/common/io/rpc/PinpointSocket.java b/src/main/java/com/nhn/pinpoint/common/io/rpc/PinpointSocket.java index a4d32016c..ae68cdb08 100644 --- a/src/main/java/com/nhn/pinpoint/common/io/rpc/PinpointSocket.java +++ b/src/main/java/com/nhn/pinpoint/common/io/rpc/PinpointSocket.java @@ -1,7 +1,9 @@ package com.nhn.pinpoint.common.io.rpc; import com.nhn.pinpoint.common.io.rpc.packet.RequestPacket; +import com.nhn.pinpoint.common.io.rpc.packet.ResponsePacket; import com.nhn.pinpoint.common.io.rpc.packet.SendPacket; +import com.nhn.pinpoint.common.io.rpc.packet.StreamCreateResultPacket; import org.jboss.netty.channel.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -12,7 +14,7 @@ import java.util.concurrent.atomic.AtomicInteger; /** * */ -public class PinpointSocket { +public class PinpointSocket extends SimpleChannelHandler { private final Logger logger = LoggerFactory.getLogger(this.getClass()); @@ -24,14 +26,14 @@ public class PinpointSocket { private final AtomicInteger state = new AtomicInteger(STATE_INIT); private Channel channel; - private SocketRequestHandler socketRequestHandler; - private StreamPacketDispatcher streamPacketDispatcher; + private SocketRequestHandler requestResponseManager; + private StreamChannelManager streamChannelManager; private long timeoutMillis = 3000; public PinpointSocket() { - this.socketRequestHandler = new SocketRequestHandler(); - this.streamPacketDispatcher = new StreamPacketDispatcher(); + this.requestResponseManager = new SocketRequestHandler(); + this.streamChannelManager = new StreamChannelManager(); } @@ -48,7 +50,7 @@ public class PinpointSocket { if (!(this.state.compareAndSet(STATE_INIT, STATE_RUN))) { throw new IllegalStateException("invalid open state:" + state.get()); } - this.streamPacketDispatcher.setChannel(channel); + this.streamChannelManager.setChannel(channel); } public void send(byte[] bytes) { @@ -91,11 +93,11 @@ public class PinpointSocket { return this.channel.write(send); } - public MessageFuture request(byte[] bytes) { + public Future request(byte[] bytes) { ensureOpen(); RequestPacket request = new RequestPacket(bytes); - final MessageFuture messageFuture = this.socketRequestHandler.register(request, this.timeoutMillis); + final DefaultFuture messageFuture = this.requestResponseManager.register(request, this.timeoutMillis); ChannelFuture write = this.channel.write(request); write.addListener(new ChannelFutureListener() { @@ -114,13 +116,38 @@ public class PinpointSocket { - public StreamChannelFuture createStreamChannel(byte[] bytes) { + public StreamChannel createStreamChannel() { ensureOpen(); - StreamChannel streamChannel = this.streamPacketDispatcher.createStreamChannel(); - streamChannel.open(bytes); - StreamChannelFuture streamChannelFuture = new StreamChannelFuture(streamChannel); - return streamChannelFuture; + StreamChannel streamChannel = this.streamChannelManager.createStreamChannelFuture(); + return streamChannel; + } + + + @Override + public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception { + final Object message = e.getMessage(); + if (message instanceof ResponsePacket) { + requestResponseManager.messageReceived((ResponsePacket) message, e.getChannel()); + return; + } + else if (message instanceof RequestPacket) { + requestResponseManager.messageReceived((RequestPacket) message, e.getChannel()); + // connector로 들어오는 request 메시지를 핸들링을 해야 함. + return; + } else if(message instanceof StreamCreateResultPacket) { + streamChannelManager.messageReceived((StreamCreateResultPacket)message, e.getChannel()); + return; + } + else { + logger.error("unexpectedMessage received:{} address:{}", message, e.getRemoteAddress()); + } + } + + @Override + public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws Exception { + logger.error("UnexpectedError happened. event:{}", e, e.getCause()); + } @@ -135,11 +162,8 @@ public class PinpointSocket { return; } // hand shake close - this.socketRequestHandler.close(); + this.requestResponseManager.close(); this.channel.close(); } - ChannelHandler getSocketRequestHandler() { - return this.socketRequestHandler; - } } diff --git a/src/main/java/com/nhn/pinpoint/common/io/rpc/PinpointSocketFactory.java b/src/main/java/com/nhn/pinpoint/common/io/rpc/PinpointSocketFactory.java index 2a60f9059..7c2775f78 100644 --- a/src/main/java/com/nhn/pinpoint/common/io/rpc/PinpointSocketFactory.java +++ b/src/main/java/com/nhn/pinpoint/common/io/rpc/PinpointSocketFactory.java @@ -74,11 +74,10 @@ public class PinpointSocketFactory { connectFuture.addListener(new ChannelFutureListener() { @Override public void operationComplete(ChannelFuture future) throws Exception { - if (future.isSuccess()) { - Channel channel = future.getChannel(); - ChannelPipeline pipeline = channel.getPipeline(); - pipeline.addLast("requestHandler", pinpointSocket.getSocketRequestHandler()); - } + // connect에 성공하든 실패하던 다 보는게 더 좋을듯하다. + Channel channel = future.getChannel(); + ChannelPipeline pipeline = channel.getPipeline(); + pipeline.addLast("logic", pinpointSocket); } }); // connectTimeout이 있어서 그냥 기다리면됨. diff --git a/src/main/java/com/nhn/pinpoint/common/io/rpc/RequestProcessor.java b/src/main/java/com/nhn/pinpoint/common/io/rpc/RequestProcessor.java index 2113e60a6..a8f273a95 100644 --- a/src/main/java/com/nhn/pinpoint/common/io/rpc/RequestProcessor.java +++ b/src/main/java/com/nhn/pinpoint/common/io/rpc/RequestProcessor.java @@ -16,13 +16,13 @@ import java.util.concurrent.atomic.AtomicInteger; /** * */ -public class RequestProcessor implements FailureHandle { +public class RequestProcessor { private final Logger logger = LoggerFactory.getLogger(this.getClass()); private final AtomicInteger requestId = new AtomicInteger(1); - private final ConcurrentMap requestMap = new ConcurrentHashMap(); + private final ConcurrentMap> requestMap = new ConcurrentHashMap>(); // Timer를 factory로 옮겨야 되나? private final HashedWheelTimer timer; @@ -36,25 +36,43 @@ public class RequestProcessor implements FailureHandle { } - public MessageFuture registerRequest(final RequestPacket request, long timeoutMillis) { + public DefaultFuture registerRequest(final RequestPacket request, long timeoutMillis) { // shutdown check final int requestId = getNextRequestId(); request.setRequestId(requestId); - final MessageFuture future = new MessageFuture(requestId, timeoutMillis); + final DefaultFuture future = new DefaultFuture(timeoutMillis); - final MessageFuture old = this.requestMap.put(requestId, future); + final Future old = this.requestMap.put(requestId, future); if (old != null) { throw new PinpointSocketException("unexpected error. old future exist:" + old + " id:" + requestId); } // future가 실패하였을 경우 requestMap에서 빠르게 지울수 있도록 핸들을 넣는다. - future.setFailureHandle(this); + FailureEventHandler removeTable = createFailureEventHandler(requestId); + future.setFailureEventHandler(removeTable); addTimeoutTask(timeoutMillis, future); return future; } - private void addTimeoutTask(long timeoutMillis, MessageFuture future) { + private FailureEventHandler createFailureEventHandler(final int requestId) { + FailureEventHandler failureEventHandler = new FailureEventHandler() { + @Override + public boolean fireFailure() { + DefaultFuture future = RequestProcessor.this.removeMessageFuture(requestId); + if (future != null) { + // 정확하게 지워짐. + return true; + } + return false; + } + }; + return failureEventHandler; + } + + + + private void addTimeoutTask(long timeoutMillis, DefaultFuture future) { try { Timeout timeout = timer.newTimeout(future, timeoutMillis, TimeUnit.MILLISECONDS); future.setTimeout(timeout); @@ -68,23 +86,24 @@ public class RequestProcessor implements FailureHandle { return this.requestId.getAndIncrement(); } - public MessageFuture removeMessageFuture(int requestId) { + public DefaultFuture removeMessageFuture(int requestId) { return this.requestMap.remove(requestId); } public void close() { final PinpointSocketException closed = new PinpointSocketException("connection closed"); + // close의 동시성 타이밍을 좀더 좋게 맞출수는 없나? final HashedWheelTimer timer = this.timer; if (timer != null) { Set stop = timer.stop(); for (Timeout timeout : stop) { - MessageFuture future = (MessageFuture)timeout.getTask(); + DefaultFuture future = (DefaultFuture)timeout.getTask(); future.setFailure(closed); } } - for (Map.Entry entry : requestMap.entrySet()) { + for (Map.Entry> entry : requestMap.entrySet()) { entry.getValue().setFailure(closed); } this.requestMap.clear(); @@ -92,8 +111,4 @@ public class RequestProcessor implements FailureHandle { } - @Override - public void handleFailure(int requestId) { - removeMessageFuture(requestId); - } } diff --git a/src/main/java/com/nhn/pinpoint/common/io/rpc/ServerMessageListener.java b/src/main/java/com/nhn/pinpoint/common/io/rpc/ServerMessageListener.java index 33afa6496..9a5afbff4 100644 --- a/src/main/java/com/nhn/pinpoint/common/io/rpc/ServerMessageListener.java +++ b/src/main/java/com/nhn/pinpoint/common/io/rpc/ServerMessageListener.java @@ -1,7 +1,6 @@ package com.nhn.pinpoint.common.io.rpc; -import com.nhn.pinpoint.common.io.rpc.packet.RequestPacket; -import com.nhn.pinpoint.common.io.rpc.packet.SendPacket; +import com.nhn.pinpoint.common.io.rpc.packet.*; import org.jboss.netty.channel.Channel; /** @@ -12,8 +11,8 @@ public interface ServerMessageListener { void handleRequest(RequestPacket requestPacket, Channel channel); - void handleStreamCreate(); - void handleStreamMessage(); - void handleStreamClosed(); + void handleStream(StreamPacket requestPacket, Channel channel); + + } diff --git a/src/main/java/com/nhn/pinpoint/common/io/rpc/SimpleSeverMessageListener.java b/src/main/java/com/nhn/pinpoint/common/io/rpc/SimpleSeverMessageListener.java index bd67c5a3f..3d53a1843 100644 --- a/src/main/java/com/nhn/pinpoint/common/io/rpc/SimpleSeverMessageListener.java +++ b/src/main/java/com/nhn/pinpoint/common/io/rpc/SimpleSeverMessageListener.java @@ -1,8 +1,6 @@ package com.nhn.pinpoint.common.io.rpc; -import com.nhn.pinpoint.common.io.rpc.packet.RequestPacket; -import com.nhn.pinpoint.common.io.rpc.packet.ResponsePacket; -import com.nhn.pinpoint.common.io.rpc.packet.SendPacket; +import com.nhn.pinpoint.common.io.rpc.packet.*; import org.jboss.netty.channel.Channel; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -26,18 +24,11 @@ public class SimpleSeverMessageListener implements ServerMessageListener { channel.write(responsePacket); } + @Override - public void handleStreamCreate() { + public void handleStream(StreamPacket streamPacket, Channel channel) { } - @Override - public void handleStreamMessage() { - } - - @Override - public void handleStreamClosed() { - - } } diff --git a/src/main/java/com/nhn/pinpoint/common/io/rpc/SocketRequestHandler.java b/src/main/java/com/nhn/pinpoint/common/io/rpc/SocketRequestHandler.java index b151f3a39..3145e8759 100644 --- a/src/main/java/com/nhn/pinpoint/common/io/rpc/SocketRequestHandler.java +++ b/src/main/java/com/nhn/pinpoint/common/io/rpc/SocketRequestHandler.java @@ -2,17 +2,14 @@ package com.nhn.pinpoint.common.io.rpc; import com.nhn.pinpoint.common.io.rpc.packet.RequestPacket; import com.nhn.pinpoint.common.io.rpc.packet.ResponsePacket; -import org.jboss.netty.channel.ChannelHandlerContext; -import org.jboss.netty.channel.ExceptionEvent; -import org.jboss.netty.channel.MessageEvent; -import org.jboss.netty.channel.SimpleChannelHandler; +import org.jboss.netty.channel.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; /** * */ -public class SocketRequestHandler extends SimpleChannelHandler { +public class SocketRequestHandler { private final Logger logger = LoggerFactory.getLogger(this.getClass()); @@ -21,44 +18,28 @@ public class SocketRequestHandler extends SimpleChannelHandler { public SocketRequestHandler() { } - @Override - public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception { - final Object message = e.getMessage(); - if (message instanceof ResponsePacket) { - final ResponsePacket responsePacket = (ResponsePacket) message; - - final int requestId = responsePacket.getRequestId(); - final MessageFuture messageFuture = requestProcessor.removeMessageFuture(requestId); - if (messageFuture == null) { - logger.warn("messageFuture not found:{}, channel:{}", responsePacket, e.getChannel()); - return; - } else { - logger.debug("responsePacket arrived packet:{}, channel:{}", responsePacket, e.getChannel()); - } - - ResponseMessage response = new ResponseMessage(); - response.setMessage(responsePacket.getPayload()); - messageFuture.setMessage(response); + public void messageReceived(ResponsePacket responsePacket, Channel channel) { + final int requestId = responsePacket.getRequestId(); + final DefaultFuture future = requestProcessor.removeMessageFuture(requestId); + if (future == null) { + logger.warn("future not found:{}, channel:{}", responsePacket, channel); return; + } else { + logger.debug("responsePacket arrived packet:{}, channel:{}", responsePacket, channel); } - else if (message instanceof RequestPacket) { - RequestPacket rp = (RequestPacket) message; - // connector로 들어오는 request 메시지를 핸들링을 해야 함. - return; - } - else { - logger.error("unexpectedMessage received:{} address:{}", message, e.getRemoteAddress()); - } + + ResponseMessage response = new ResponseMessage(); + response.setMessage(responsePacket.getPayload()); + future.setObject(response); } - @Override - public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws Exception { - logger.error("UnexpectedError happened. event:{}", e, e.getCause()); - + public void messageReceived(RequestPacket requestPacket, Channel channel) { + logger.error("unexpectedMessage received:{} address:{}", requestPacket, channel.getRemoteAddress()); } - public MessageFuture register(RequestPacket requestPacket, long timeoutMillis) { + + public DefaultFuture register(RequestPacket requestPacket, long timeoutMillis) { return this.requestProcessor.registerRequest(requestPacket, timeoutMillis); } diff --git a/src/main/java/com/nhn/pinpoint/common/io/rpc/StreamChannel.java b/src/main/java/com/nhn/pinpoint/common/io/rpc/StreamChannel.java index d6df86ca1..fbb21cfdc 100644 --- a/src/main/java/com/nhn/pinpoint/common/io/rpc/StreamChannel.java +++ b/src/main/java/com/nhn/pinpoint/common/io/rpc/StreamChannel.java @@ -1,6 +1,9 @@ package com.nhn.pinpoint.common.io.rpc; import com.nhn.pinpoint.common.io.rpc.packet.StreamCreatePacket; +import com.nhn.pinpoint.common.io.rpc.packet.StreamPacket; +import org.jboss.netty.channel.ChannelFuture; +import org.jboss.netty.channel.ChannelFutureListener; import java.util.concurrent.atomic.AtomicInteger; @@ -8,16 +11,24 @@ import java.util.concurrent.atomic.AtomicInteger; * */ public class StreamChannel { + private static final int NONE = 0; + // OPEN 호출 + private static final int OPEN = 1; + // OPEN 결과 대기 + private static final int OPEN_AWAIT = 2; + // 동작중 + private static final int RUN = 3; + // 닫힘 + private static final int CLOSED = 4; - private AtomicInteger state = new AtomicInteger(0); - + private AtomicInteger state = new AtomicInteger(NONE); private final int channelId; - private StreamPacketDispatcher streamPacketDispatcher; + private StreamChannelManager streamChannelManager; - private StreamChannelFuture openLatch; +// private DefaultFuture openLatch = new DefaultFuture(0); public StreamChannel(int channelId) { this.channelId = channelId; @@ -27,33 +38,75 @@ public class StreamChannel { return channelId; } - public StreamChannelFuture open(byte[] bytes) { + public Future open(byte[] bytes) { + if (!state.compareAndSet(NONE, OPEN)) { + throw new IllegalStateException("invalid state"); + } StreamCreatePacket streamCreatePacket = new StreamCreatePacket(channelId, bytes); - this.streamPacketDispatcher.writeStreamPacket(streamCreatePacket); - openLatch = new StreamChannelFuture(this); - return openLatch; + final DefaultFuture future = new DefaultFuture(); + future.setFailureEventHandler(new FailureEventHandler() { + @Override + public boolean fireFailure() { + streamChannelManager.closeChannel(channelId); + return false; + } + }); + ChannelFuture channelFuture = this.streamChannelManager.writeStreamPacket(streamCreatePacket); + channelFuture.addListener(new ChannelFutureListener() { + @Override + public void operationComplete(ChannelFuture future) throws Exception { + if (!future.isSuccess()) { + future.setFailure(future.getCause()); + } + } + }); + + + if (!state.compareAndSet(OPEN, OPEN_AWAIT)) { + throw new IllegalStateException("invalid state"); + } + return future; } public void setStreamResponseListener() { } - public void receiveStreamResponse(byte[] stream) { - openLatch.open(); + public boolean receiveStreamPacket(StreamPacket packet) { + + return true; } public synchronized void close() { - StreamPacketDispatcher streamPacketDispatcher = this.streamPacketDispatcher; - if (streamPacketDispatcher != null) { - streamPacketDispatcher.closeChannel(channelId); - this.streamPacketDispatcher = null; + StreamChannelManager streamChannelManager = this.streamChannelManager; + if (streamChannelManager != null) { + streamChannelManager.closeChannel(channelId); + this.streamChannelManager = null; } } - public void setStreamPacketDispatcher(StreamPacketDispatcher streamPacketDispatcher) { - this.streamPacketDispatcher = streamPacketDispatcher; + public void setStreamChannelManager(StreamChannelManager streamChannelManager) { + this.streamChannelManager = streamChannelManager; + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + + StreamChannel that = (StreamChannel) o; + + if (channelId != that.channelId) return false; + + return true; + } + + @Override + public int hashCode() { + return channelId; } } + diff --git a/src/main/java/com/nhn/pinpoint/common/io/rpc/StreamChannelFuture.java b/src/main/java/com/nhn/pinpoint/common/io/rpc/StreamChannelFuture.java deleted file mode 100644 index d8732e4b2..000000000 --- a/src/main/java/com/nhn/pinpoint/common/io/rpc/StreamChannelFuture.java +++ /dev/null @@ -1,16 +0,0 @@ -package com.nhn.pinpoint.common.io.rpc; - -/** - * - */ -public class StreamChannelFuture { - private StreamChannel streamChannel; - public StreamChannelFuture(StreamChannel streamChannel) { - this.streamChannel = streamChannel; - } - - public void open() { - - - } -} diff --git a/src/main/java/com/nhn/pinpoint/common/io/rpc/StreamPacketDispatcher.java b/src/main/java/com/nhn/pinpoint/common/io/rpc/StreamChannelManager.java similarity index 57% rename from src/main/java/com/nhn/pinpoint/common/io/rpc/StreamPacketDispatcher.java rename to src/main/java/com/nhn/pinpoint/common/io/rpc/StreamChannelManager.java index a934f7ed2..b6a5e6eb5 100644 --- a/src/main/java/com/nhn/pinpoint/common/io/rpc/StreamPacketDispatcher.java +++ b/src/main/java/com/nhn/pinpoint/common/io/rpc/StreamChannelManager.java @@ -1,7 +1,9 @@ package com.nhn.pinpoint.common.io.rpc; +import com.nhn.pinpoint.common.io.rpc.packet.StreamCreateResultPacket; import com.nhn.pinpoint.common.io.rpc.packet.StreamPacket; import org.jboss.netty.channel.Channel; +import org.jboss.netty.channel.ChannelFuture; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -12,7 +14,7 @@ import java.util.concurrent.atomic.AtomicInteger; /** * */ -public class StreamPacketDispatcher { +public class StreamChannelManager { private Logger logger = LoggerFactory.getLogger(this.getClass()); @@ -21,17 +23,18 @@ public class StreamPacketDispatcher { private final ConcurrentMap channelMap = new ConcurrentHashMap(); - public StreamChannel createStreamChannel() { + public StreamChannel createStreamChannelFuture() { + final int channelId = allocateChannelId(); StreamChannel streamChannel = new StreamChannel(channelId); StreamChannel old = channelMap.put(channelId, streamChannel); if (old != null) { - throw new IllegalStateException("already channelId exist:" + channelId + " streamChannel:" + old); + throw new PinpointSocketException("already channelId exist:" + channelId + " streamChannel:" + old); } - // handler을 붙여서 리턴. - streamChannel.setStreamPacketDispatcher(this); + // handle을 붙여서 리턴. + streamChannel.setStreamChannelManager(this); return streamChannel; } @@ -46,8 +49,11 @@ public class StreamPacketDispatcher { logger.warn("streamChannel not found channelId:{}", packet.getChannelId()); return false; } -// streamChannel.receiveStreamResponse(packet); - return true; + return streamChannel.receiveStreamPacket(packet); + } + + public StreamChannel findStreamChannel(int channelId) { + return this.channelMap.get(channelId); } public boolean closeChannel(int channelId) { @@ -55,11 +61,21 @@ public class StreamPacketDispatcher { return remove != null; } - public void writeStreamPacket(StreamPacket packet) { - channel.write(packet); + public ChannelFuture writeStreamPacket(StreamPacket packet) { + // connector이 close상태 체크가 필요함. + // ex : ensureOpen() + + return channel.write(packet); } public void setChannel(Channel channel) { this.channel = channel; } + + public void messageReceived(StreamCreateResultPacket streamCreateResult, Channel channel) { + int channelId = streamCreateResult.getChannelId(); + StreamChannel streamChannel = findStreamChannel(channelId); + streamChannel.receiveStreamPacket(streamCreateResult); + + } } diff --git a/src/test/java/com/nhn/pinpoint/common/io/rpc/MessageFutureTest.java b/src/test/java/com/nhn/pinpoint/common/io/rpc/FutureTest.java similarity index 57% rename from src/test/java/com/nhn/pinpoint/common/io/rpc/MessageFutureTest.java rename to src/test/java/com/nhn/pinpoint/common/io/rpc/FutureTest.java index 72f4235c4..bdec3719e 100644 --- a/src/test/java/com/nhn/pinpoint/common/io/rpc/MessageFutureTest.java +++ b/src/test/java/com/nhn/pinpoint/common/io/rpc/FutureTest.java @@ -3,5 +3,5 @@ package com.nhn.pinpoint.common.io.rpc; /** * */ -public class MessageFutureTest { +public class FutureTest { } diff --git a/src/test/java/com/nhn/pinpoint/common/io/rpc/PinpointSocketFactoryTest.java b/src/test/java/com/nhn/pinpoint/common/io/rpc/PinpointSocketFactoryTest.java index 72b5377b6..199ab89d2 100644 --- a/src/test/java/com/nhn/pinpoint/common/io/rpc/PinpointSocketFactoryTest.java +++ b/src/test/java/com/nhn/pinpoint/common/io/rpc/PinpointSocketFactoryTest.java @@ -40,14 +40,17 @@ public class PinpointSocketFactoryTest { socket.send(new byte[20]); byte[] bytes = new byte[10]; bytes[0] = 1; - MessageFuture request = socket.request(bytes); + Future request = socket.request(bytes); request.await(); - Message message = request.getMessage(); + ResponseMessage message = request.getObject(); Assert.assertArrayEquals(message.getMessage(), bytes); socket.sendSync(new byte[20]); -// StreamChannelFuture streamChannel = socket.createStreamChannel(); + StreamChannel streamChannel = socket.createStreamChannel(); + Future open = streamChannel.open(new byte[10]); + open.await(); + socket.close(); } finally { diff --git a/src/test/java/com/nhn/pinpoint/common/io/rpc/RequestProcessorTest.java b/src/test/java/com/nhn/pinpoint/common/io/rpc/RequestProcessorTest.java index 1d451b299..8af1466b6 100644 --- a/src/test/java/com/nhn/pinpoint/common/io/rpc/RequestProcessorTest.java +++ b/src/test/java/com/nhn/pinpoint/common/io/rpc/RequestProcessorTest.java @@ -19,13 +19,13 @@ public class RequestProcessorTest { RequestProcessor requestProcessor = new RequestProcessor(10); try { RequestPacket packet = new RequestPacket(new byte[0]); - MessageFuture messageFuture = requestProcessor.registerRequest(packet, 50); + Future future = requestProcessor.registerRequest(packet, 50); Thread.sleep(200); - Assert.assertTrue(messageFuture.isReady()); - Assert.assertFalse(messageFuture.isSuccess()); - Assert.assertTrue(messageFuture.getCause().getMessage().contains("timeout")); - logger.debug(messageFuture.getCause().getMessage()); + Assert.assertTrue(future.isReady()); + Assert.assertFalse(future.isSuccess()); + Assert.assertTrue(future.getCause().getMessage().contains("timeout")); + logger.debug(future.getCause().getMessage()); } finally { requestProcessor.close(); } @@ -37,11 +37,11 @@ public class RequestProcessorTest { RequestProcessor requestProcessor = new RequestProcessor(10); try { RequestPacket packet = new RequestPacket(1, new byte[0]); - MessageFuture messageFuture = requestProcessor.registerRequest(packet, 2000); + DefaultFuture future = requestProcessor.registerRequest(packet, 2000); - messageFuture.setFailure(new RuntimeException()); + future.setFailure(new RuntimeException()); - MessageFuture nullFuture = requestProcessor.removeMessageFuture(packet.getRequestId()); + Future nullFuture = requestProcessor.removeMessageFuture(packet.getRequestId()); Assert.assertNull(nullFuture);