mirror of
https://github.com/wahyd4/pinpoint.git
synced 2026-08-16 16:28:48 +10:00
[강운덕] [LUCYSUS-1744] stream 채널관리 기능 개발
git-svn-id: http://svn.bds.nhncorp.com/pe/hippo-commons/trunk@1952 84d0f5b1-2673-498c-a247-62c4ff18d310
This commit is contained in:
Woonduk Kang
+30
-25
@@ -8,60 +8,62 @@ import org.slf4j.LoggerFactory;
|
||||
/**
|
||||
*
|
||||
*/
|
||||
public class MessageFuture implements TimerTask {
|
||||
public class DefaultFuture<T> implements TimerTask, Future<T> {
|
||||
|
||||
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<T> 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<T> 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<T> 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<T> 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;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,9 @@
|
||||
package com.nhn.pinpoint.common.io.rpc;
|
||||
|
||||
/**
|
||||
*
|
||||
*/
|
||||
public interface FailureEventHandler {
|
||||
|
||||
boolean fireFailure();
|
||||
}
|
||||
@@ -1,9 +0,0 @@
|
||||
package com.nhn.pinpoint.common.io.rpc;
|
||||
|
||||
/**
|
||||
*
|
||||
*/
|
||||
public interface FailureHandle {
|
||||
|
||||
void handleFailure(int requestId);
|
||||
}
|
||||
@@ -0,0 +1,20 @@
|
||||
package com.nhn.pinpoint.common.io.rpc;
|
||||
|
||||
/**
|
||||
*
|
||||
*/
|
||||
public interface Future<T> {
|
||||
T getObject();
|
||||
|
||||
Throwable getCause();
|
||||
|
||||
boolean isReady();
|
||||
|
||||
boolean isSuccess();
|
||||
|
||||
boolean setListener(FutureListener<T> listener);
|
||||
|
||||
boolean await(long timeoutMillis);
|
||||
|
||||
boolean await();
|
||||
}
|
||||
@@ -0,0 +1,8 @@
|
||||
package com.nhn.pinpoint.common.io.rpc;
|
||||
|
||||
/**
|
||||
*
|
||||
*/
|
||||
public interface FutureListener<T> {
|
||||
void onComplete(Future<T> future);
|
||||
}
|
||||
@@ -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);
|
||||
}
|
||||
@@ -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());
|
||||
}
|
||||
|
||||
|
||||
@@ -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<ResponseMessage> request(byte[] bytes) {
|
||||
ensureOpen();
|
||||
|
||||
RequestPacket request = new RequestPacket(bytes);
|
||||
final MessageFuture messageFuture = this.socketRequestHandler.register(request, this.timeoutMillis);
|
||||
final DefaultFuture<ResponseMessage> 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;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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이 있어서 그냥 기다리면됨.
|
||||
|
||||
@@ -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<Integer, MessageFuture> requestMap = new ConcurrentHashMap<Integer, MessageFuture>();
|
||||
private final ConcurrentMap<Integer, DefaultFuture<ResponseMessage>> requestMap = new ConcurrentHashMap<Integer, DefaultFuture<ResponseMessage>>();
|
||||
// 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<ResponseMessage> registerRequest(final RequestPacket request, long timeoutMillis) {
|
||||
// shutdown check
|
||||
final int requestId = getNextRequestId();
|
||||
request.setRequestId(requestId);
|
||||
|
||||
final MessageFuture future = new MessageFuture(requestId, timeoutMillis);
|
||||
final DefaultFuture<ResponseMessage> future = new DefaultFuture<ResponseMessage>(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<ResponseMessage> 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<ResponseMessage> 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<Timeout> stop = timer.stop();
|
||||
for (Timeout timeout : stop) {
|
||||
MessageFuture future = (MessageFuture)timeout.getTask();
|
||||
DefaultFuture future = (DefaultFuture)timeout.getTask();
|
||||
future.setFailure(closed);
|
||||
}
|
||||
}
|
||||
|
||||
for (Map.Entry<Integer, MessageFuture> entry : requestMap.entrySet()) {
|
||||
for (Map.Entry<Integer, DefaultFuture<ResponseMessage>> 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);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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);
|
||||
|
||||
|
||||
|
||||
}
|
||||
|
||||
@@ -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() {
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<ResponseMessage> 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<ResponseMessage> register(RequestPacket requestPacket, long timeoutMillis) {
|
||||
return this.requestProcessor.registerRequest(requestPacket, timeoutMillis);
|
||||
}
|
||||
|
||||
|
||||
@@ -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<StreamChannel> openLatch = new DefaultFuture<StreamChannel>(0);
|
||||
|
||||
public StreamChannel(int channelId) {
|
||||
this.channelId = channelId;
|
||||
@@ -27,33 +38,75 @@ public class StreamChannel {
|
||||
return channelId;
|
||||
}
|
||||
|
||||
public StreamChannelFuture open(byte[] bytes) {
|
||||
public Future<StreamChannel> 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<StreamChannel> future = new DefaultFuture<StreamChannel>();
|
||||
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;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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() {
|
||||
|
||||
|
||||
}
|
||||
}
|
||||
+25
-9
@@ -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<Integer, StreamChannel> channelMap = new ConcurrentHashMap<Integer, StreamChannel>();
|
||||
|
||||
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);
|
||||
|
||||
}
|
||||
}
|
||||
+1
-1
@@ -3,5 +3,5 @@ package com.nhn.pinpoint.common.io.rpc;
|
||||
/**
|
||||
*
|
||||
*/
|
||||
public class MessageFutureTest {
|
||||
public class FutureTest {
|
||||
}
|
||||
@@ -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<ResponseMessage> 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<StreamChannel> open = streamChannel.open(new byte[10]);
|
||||
open.await();
|
||||
|
||||
|
||||
socket.close();
|
||||
} finally {
|
||||
|
||||
@@ -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);
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user