diff --git a/pom.xml b/pom.xml
index ede0210..007532b 100644
--- a/pom.xml
+++ b/pom.xml
@@ -38,10 +38,10 @@
maven-compiler-plugin
- 2.3.2
+ 3.3
- 1.5
- 1.5
+ 1.8
+ 1.8
@@ -91,27 +91,18 @@
-
-
- repository.jboss.org
- https://repository.jboss.org/nexus/content/groups/public/
-
- false
-
-
-
org.msgpack
msgpack
- 0.6.6
+ 0.6.12
compile
- org.jboss.netty
- netty
- 3.2.1.Final
+ io.netty
+ netty-all
+ 4.0.28.Final
javax.servlet
diff --git a/src/main/java/org/msgpack/rpc/Request.java b/src/main/java/org/msgpack/rpc/Request.java
index a0007e1..f82d7f1 100644
--- a/src/main/java/org/msgpack/rpc/Request.java
+++ b/src/main/java/org/msgpack/rpc/Request.java
@@ -15,15 +15,32 @@
// See the License for the specific language governing permissions and
// limitations under the License.
//
+/*
+* Copyright (C) 2014-2015 Information Analysis Laboratory, NICT
+*
+* RaSC is free software: you can redistribute it and/or modify it
+* under the terms of the GNU Lesser General Public License as published by
+* the Free Software Foundation, either version 2.1 of the License, or (at
+* your option) any later version.
+*
+* RaSC is distributed in the hope that it will be useful, but
+* WITHOUT ANY WARRANTY; without even the implied warranty of
+* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser
+* General Public License for more details.
+*
+* You should have received a copy of the GNU Lesser General Public License
+* along with this program. If not, see {
- private MessageSendable channel; // TODO #SF synchronized?
- private int msgid;
+ protected MessageSendable channel; // TODO #SF synchronized?
+ protected int msgid;
private String method;
private Value args;
diff --git a/src/main/java/org/msgpack/rpc/Server.java b/src/main/java/org/msgpack/rpc/Server.java
index ba027d5..bf178dc 100644
--- a/src/main/java/org/msgpack/rpc/Server.java
+++ b/src/main/java/org/msgpack/rpc/Server.java
@@ -15,27 +15,41 @@
// See the License for the specific language governing permissions and
// limitations under the License.
//
+/*
+* Copyright (C) 2014-2015 Information Analysis Laboratory, NICT
+*
+* RaSC is free software: you can redistribute it and/or modify it
+* under the terms of the GNU Lesser General Public License as published by
+* the Free Software Foundation, either version 2.1 of the License, or (at
+* your option) any later version.
+*
+* RaSC is distributed in the hope that it will be useful, but
+* WITHOUT ANY WARRANTY; without even the implied warranty of
+* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser
+* General Public License for more details.
+*
+* You should have received a copy of the GNU Lesser General Public License
+* along with this program. If not, see reqtable = new HashMap();
+ protected AtomicInteger seqid = new AtomicInteger(0); // FIXME rand()?
+ protected Map reqtable = new HashMap();
Session(Address address, ClientConfig config, EventLoop loop) {
this(address,config,loop,new Reflect(loop.getMessagePack()));
@@ -141,16 +159,14 @@ void closeSession() {
}
public void transportConnectFailed() { // FIXME error rseult
- /*
synchronized(reqtable) {
for(Map.Entry pair : reqtable.entrySet()) {
// FIXME
FutureImpl f = pair.getValue();
- f.setResult(null,null);
+ f.setResult(null,ValueFactory.createRawValue("Fail to connect"));
}
reqtable.clear();
}
- */
}
public void onResponse(int msgid, Value result, Value error) {
@@ -183,4 +199,15 @@ void stepTimeout() {
f.setResult(null, ValueFactory.createRawValue("timedout"));
}
}
+
+ public void transportError(String msg){
+ synchronized(reqtable) {
+ for(Map.Entry pair : reqtable.entrySet()) {
+ // FIXME
+ FutureImpl f = pair.getValue();
+ f.setResult(null,ValueFactory.createRawValue(msg));
+ }
+ reqtable.clear();
+ }
+ }
}
diff --git a/src/main/java/org/msgpack/rpc/config/ClientConfig.java b/src/main/java/org/msgpack/rpc/config/ClientConfig.java
index 1df431f..695e1d3 100644
--- a/src/main/java/org/msgpack/rpc/config/ClientConfig.java
+++ b/src/main/java/org/msgpack/rpc/config/ClientConfig.java
@@ -15,13 +15,32 @@
// See the License for the specific language governing permissions and
// limitations under the License.
//
+/*
+* Copyright (C) 2014-2015 Information Analysis Laboratory, NICT
+*
+* RaSC is free software: you can redistribute it and/or modify it
+* under the terms of the GNU Lesser General Public License as published by
+* the Free Software Foundation, either version 2.1 of the License, or (at
+* your option) any later version.
+*
+* RaSC is distributed in the hope that it will be useful, but
+* WITHOUT ANY WARRANTY; without even the implied warranty of
+* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser
+* General Public License for more details.
+*
+* You should have received a copy of the GNU Lesser General Public License
+* along with this program. If not, see options = new HashMap();
+ private Map,Object> options = new ConcurrentHashMap<>();
protected int requestTimeout = 30; // FIXME default timeout time
public void setRequestTimeout(int sec) {
@@ -32,15 +51,16 @@ public int getRequestTimeout() {
return this.requestTimeout;
}
- public Object getOption(String key) {
- return options.get(key);
+ @SuppressWarnings("unchecked")
+ public T getOption(ChannelOption key) {
+ return (T)options.get(key);
}
- public Map getOptions() {
+ public Map,Object> getOptions() {
return options;
}
- public void setOption(String key, Object value) {
+ public void setOption(ChannelOption key, T value) {
options.put(key, value);
}
}
diff --git a/src/main/java/org/msgpack/rpc/extension/socks/SocksProxyHandler.java b/src/main/java/org/msgpack/rpc/extension/socks/SocksProxyHandler.java
new file mode 100644
index 0000000..af8a426
--- /dev/null
+++ b/src/main/java/org/msgpack/rpc/extension/socks/SocksProxyHandler.java
@@ -0,0 +1,110 @@
+/*
+* Copyright (C) 2014-2015 Information Analysis Laboratory, NICT
+*
+* RaSC is free software: you can redistribute it and/or modify it
+* under the terms of the GNU Lesser General Public License as published by
+* the Free Software Foundation, either version 2.1 of the License, or (at
+* your option) any later version.
+*
+* RaSC is distributed in the hope that it will be useful, but
+* WITHOUT ANY WARRANTY; without even the implied warranty of
+* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser
+* General Public License for more details.
+*
+* You should have received a copy of the GNU Lesser General Public License
+* along with this program. If not, see {
+
+ private Session session = null;
+ private NettyTcpClientTransport clientTransport = null;
+
+ public SocksProxyHandler() {
+ super();
+ }
+
+ public SocksProxyHandler(Session session,NettyTcpClientTransport clientTransport) {
+ super();
+ this.session = session;
+ this.clientTransport = clientTransport;
+ }
+
+ @Override
+ protected void channelRead0(ChannelHandlerContext ctx, SocksResponse response) throws Exception {
+
+ SocksResponseType rt = response.responseType();
+
+ switch (rt) {
+
+ case INIT:
+ ctx.pipeline().addFirst("socks-cmd-decoder", new SocksCmdResponseDecoder());
+ InetSocketAddress addr = ((IPAddress) session.getAddress()).getInetSocketAddress();
+ SocksCmdRequest cmdSocks = new SocksCmdRequest(SocksCmdType.CONNECT, SocksAddressType.DOMAIN, addr.getHostName(), addr.getPort());
+ ctx.writeAndFlush(cmdSocks);
+ break;
+
+ case AUTH:
+ break;
+ case CMD:
+ SocksCmdResponse scr = (SocksCmdResponse) response;
+ ctx.pipeline().remove(this);
+ ctx.pipeline().remove("socks-encode");
+ if (scr.cmdStatus() != SocksCmdStatus.SUCCESS) {
+ throw new ChannelException("Socks faild.");
+ }
+ clientTransport.onSocksConnected(ctx.channel());
+ break;
+ case UNKNOWN:
+ default:
+ throw new ChannelException("No support Socks Command.");
+ }
+ }
+
+ @Override
+ public void channelActive(ChannelHandlerContext ctx) throws Exception {
+ List lstAuth = new ArrayList<>();
+ lstAuth.add(SocksAuthScheme.NO_AUTH);
+ SocksInitRequest si = new SocksInitRequest(lstAuth);
+ ctx.writeAndFlush(si);
+ }
+
+ @Override
+ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+ cause.printStackTrace();
+ clientTransport.onError(ctx.channel(), cause.toString());
+ }
+
+}
diff --git a/src/main/java/org/msgpack/rpc/loop/EventLoop.java b/src/main/java/org/msgpack/rpc/loop/EventLoop.java
index 7a500b7..1d8fc16 100644
--- a/src/main/java/org/msgpack/rpc/loop/EventLoop.java
+++ b/src/main/java/org/msgpack/rpc/loop/EventLoop.java
@@ -15,24 +15,41 @@
// See the License for the specific language governing permissions and
// limitations under the License.
//
+/*
+* Copyright (C) 2014-2015 Information Analysis Laboratory, NICT
+*
+* RaSC is free software: you can redistribute it and/or modify it
+* under the terms of the GNU Lesser General Public License as published by
+* the Free Software Foundation, either version 2.1 of the License, or (at
+* your option) any later version.
+*
+* RaSC is distributed in the hope that it will be useful, but
+* WITHOUT ANY WARRANTY; without even the implied warranty of
+* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser
+* General Public License for more details.
+*
+* You should have received a copy of the GNU Lesser General Public License
+* along with this program. If not, see {
private RpcMessageHandler handler;
private ChannelAdaptor adaptor;
+ private NettyTcpClientTransport clientTransport = null;
MessageHandler(RpcMessageHandler handler) {
this.handler = handler;
}
- @Override
- public void channelOpen(ChannelHandlerContext ctx, ChannelStateEvent e)
- throws Exception {
- this.adaptor = new ChannelAdaptor(e.getChannel());
- ctx.sendUpstream(e);
- }
+ public MessageHandler(RpcMessageHandler handler,NettyTcpClientTransport clientTransport) {
+ this.handler = handler;
+ this.clientTransport = clientTransport;
- @Override
- public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) {
- Object m = e.getMessage();
- if (!(m instanceof Value)) {
- ctx.sendUpstream(e);
- return;
- }
+ }
- Value msg = (Value) m;
+ @Override
+ protected void channelRead0(ChannelHandlerContext ctx, Value msg) throws Exception {
handler.handleMessage(adaptor, msg);
- }
+ }
+
+ @Override
+ public void channelActive(ChannelHandlerContext ctx) throws Exception {
+ this.adaptor = new ChannelAdaptor(ctx);
+ clientTransport.onConnected(ctx.channel());
+ }
+
+
+ @Override
+ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+ if(Objects.nonNull(clientTransport)){
+ clientTransport.onError(ctx.channel(), cause.getMessage());
+ }
+ }
+
}
diff --git a/src/main/java/org/msgpack/rpc/loop/netty/MessagePackDecoder.java b/src/main/java/org/msgpack/rpc/loop/netty/MessagePackDecoder.java
index f977328..ab0559c 100644
--- a/src/main/java/org/msgpack/rpc/loop/netty/MessagePackDecoder.java
+++ b/src/main/java/org/msgpack/rpc/loop/netty/MessagePackDecoder.java
@@ -15,17 +15,36 @@
// See the License for the specific language governing permissions and
// limitations under the License.
//
+/*
+* Copyright (C) 2014-2015 Information Analysis Laboratory, NICT
+*
+* RaSC is free software: you can redistribute it and/or modify it
+* under the terms of the GNU Lesser General Public License as published by
+* the Free Software Foundation, either version 2.1 of the License, or (at
+* your option) any later version.
+*
+* RaSC is distributed in the hope that it will be useful, but
+* WITHOUT ANY WARRANTY; without even the implied warranty of
+* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser
+* General Public License for more details.
+*
+* You should have received a copy of the GNU Lesser General Public License
+* along with this program. If not, see result) throws Exception {
+ // TODO 自動生成されたメソッド・スタブ
- ByteBuffer buffer = source.toByteBuffer();
+
+ ByteBuffer buffer = msg.nioBuffer();
if (!buffer.hasRemaining()) {
- return null;
+ return;
}
byte[] bytes = buffer.array(); // FIXME buffer must has array
@@ -53,12 +69,6 @@ protected Object decode(ChannelHandlerContext ctx, Channel channel,
int length = buffer.arrayOffset() + buffer.limit();
Value v = messagePack.read(bytes, offset, length);
- return v;
-
- // TODO MessagePack.unpack()
- /*
- * Unpacker pac = new Unpacker(); pac.wrap(bytes, offset, length);
- * return pac.unpackObject();
- */
- }
+ result.add(v);
+ }
}
diff --git a/src/main/java/org/msgpack/rpc/loop/netty/MessagePackEncoder.java b/src/main/java/org/msgpack/rpc/loop/netty/MessagePackEncoder.java
index 2a8cc27..c0ec8eb 100644
--- a/src/main/java/org/msgpack/rpc/loop/netty/MessagePackEncoder.java
+++ b/src/main/java/org/msgpack/rpc/loop/netty/MessagePackEncoder.java
@@ -15,17 +15,31 @@
// See the License for the specific language governing permissions and
// limitations under the License.
//
+/*
+* Copyright (C) 2014-2015 Information Analysis Laboratory, NICT
+*
+* RaSC is free software: you can redistribute it and/or modify it
+* under the terms of the GNU Lesser General Public License as published by
+* the Free Software Foundation, either version 2.1 of the License, or (at
+* your option) any later version.
+*
+* RaSC is distributed in the hope that it will be useful, but
+* WITHOUT ANY WARRANTY; without even the implied warranty of
+* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser
+* General Public License for more details.
+*
+* You should have received a copy of the GNU Lesser General Public License
+* along with this program. If not, see {
private final int estimatedLength;
private MessagePack messagePack;
@@ -39,21 +53,11 @@ public MessagePackEncoder(int estimatedLength, MessagePack messagePack) {
this.messagePack = messagePack;
}
- @Override
- protected Object encode(ChannelHandlerContext ctx, Channel channel,
- Object msg) throws Exception {
- if (msg instanceof ChannelBuffer) {
- return msg;
- }
+ @Override
+ protected void encode(io.netty.channel.ChannelHandlerContext ctx, Object msg, ByteBuf paramByteBuf) throws Exception {
- ChannelBufferOutputStream out = new ChannelBufferOutputStream(
- ChannelBuffers.dynamicBuffer(estimatedLength, ctx.getChannel()
- .getConfig().getBufferFactory()));
+ byte[] buff = messagePack.write(msg);
+ paramByteBuf.writeBytes(buff);
- // MessagePack.pack(out, msg);
- messagePack.write(out, msg);
-
- ChannelBuffer result = out.buffer();
- return result;
- }
+ }
}
diff --git a/src/main/java/org/msgpack/rpc/loop/netty/MessagePackStreamDecoder.java b/src/main/java/org/msgpack/rpc/loop/netty/MessagePackStreamDecoder.java
index edfdb7e..baf12d5 100644
--- a/src/main/java/org/msgpack/rpc/loop/netty/MessagePackStreamDecoder.java
+++ b/src/main/java/org/msgpack/rpc/loop/netty/MessagePackStreamDecoder.java
@@ -15,53 +15,63 @@
// See the License for the specific language governing permissions and
// limitations under the License.
//
+/*
+* Copyright (C) 2014-2015 Information Analysis Laboratory, NICT
+*
+* RaSC is free software: you can redistribute it and/or modify it
+* under the terms of the GNU Lesser General Public License as published by
+* the Free Software Foundation, either version 2.1 of the License, or (at
+* your option) any later version.
+*
+* RaSC is distributed in the hope that it will be useful, but
+* WITHOUT ANY WARRANTY; without even the implied warranty of
+* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser
+* General Public License for more details.
+*
+* You should have received a copy of the GNU Lesser General Public License
+* along with this program. If not, see paramList) throws Exception {
- @Override
- protected Object decode(ChannelHandlerContext ctx, Channel channel,
- ChannelBuffer source) throws Exception {
- // TODO #MN will modify the body with MessagePackBufferUnpacker.
- ByteBuffer buffer = source.toByteBuffer();
- if (!buffer.hasRemaining()) {
- return null;
- }
- source.markReaderIndex();
+ if (msg.isReadable()) {
+ ByteBuffer buffer = msg.nioBuffer();
+ Unpacker unpacker = msgpack.createBufferUnpacker(buffer);
+ int lastPos = 0;
+ try {
+ while (buffer.position() < buffer.limit()) {
+ Value v = unpacker.readValue();
+ paramList.add(v);
+ lastPos = buffer.position();
+ }
+ msg.skipBytes(lastPos);
+ } catch (EOFException e) {
+ msg.skipBytes(lastPos);
+ }
+ }
+ }
- byte[] bytes = buffer.array(); // FIXME buffer must has array
- int offset = buffer.arrayOffset() + buffer.position();
- int length = buffer.arrayOffset() + buffer.limit();
- ByteArrayInputStream stream = new ByteArrayInputStream(bytes, offset,
- length);
- int startAvailable = stream.available();
- try{
- Unpacker unpacker = msgpack.createUnpacker(stream);
- Value v = unpacker.readValue();
- source.skipBytes(startAvailable - stream.available());
- return v;
- }catch( EOFException e ){
- // not enough buffers.
- // So retry reading
- source.resetReaderIndex();
- return null;
- }
- }
}
diff --git a/src/main/java/org/msgpack/rpc/loop/netty/NettyEventLoop.java b/src/main/java/org/msgpack/rpc/loop/netty/NettyEventLoop.java
index cf71a25..8c42e67 100644
--- a/src/main/java/org/msgpack/rpc/loop/netty/NettyEventLoop.java
+++ b/src/main/java/org/msgpack/rpc/loop/netty/NettyEventLoop.java
@@ -15,58 +15,57 @@
// See the License for the specific language governing permissions and
// limitations under the License.
//
+/*
+* Copyright (C) 2014-2015 Information Analysis Laboratory, NICT
+*
+* RaSC is free software: you can redistribute it and/or modify it
+* under the terms of the GNU Lesser General Public License as published by
+* the Free Software Foundation, either version 2.1 of the License, or (at
+* your option) any later version.
+*
+* RaSC is distributed in the hope that it will be useful, but
+* WITHOUT ANY WARRANTY; without even the implied warranty of
+* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser
+* General Public License for more details.
+*
+* You should have received a copy of the GNU Lesser General Public License
+* along with this program. If not, see {
+ private static final InternalLogger LOG = InternalLoggerFactory.getInstance(NettyTcpClientTransport.class);
+ private final Bootstrap bootstrap;
+ private final EventLoopGroup group = new NioEventLoopGroup(maxChannel);
+ private RpcMessageHandler handler = null;
+ private NettyEventLoop eventLoop = null;
+
+ NettyTcpClientTransport(TcpClientConfig config, Session session, NettyEventLoop loop, Class rpcHandlerClass) {
+ super(config, session);
+
+ try {
+ handler = rpcHandlerClass.getConstructor(Session.class).newInstance(session);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+
+ // handler = new RpcMessageHandlerEx(session);
+ eventLoop = loop;
+ bootstrap = new Bootstrap().group(group);
+ bootstrap.channel(NioSocketChannel.class);
+ final NettyTcpClientTransport trans = this;
+ bootstrap.handler(new ChannelInitializer() {
+
+ @Override
+ protected void initChannel(SocketChannel ch) throws Exception {
+ ChannelPipeline p = ch.pipeline();
+ if (isSocks) {
+ p.addFirst("socks-handler", new SocksProxyHandler(session, trans));
+ p.addFirst("socks-encode", new SocksMessageEncoder());
+ p.addFirst("socks-decode", new SocksInitResponseDecoder());
+ }
+ p.addLast("msgpack-decode-stream", new MessagePackStreamDecoder(eventLoop.getMessagePack()));
+ p.addLast("msgpack-encode", new MessagePackEncoder(eventLoop.getMessagePack()));
+ p.addLast("message", new MessageHandler(handler, trans));
+ }
+
+ });
+ bootstrap.option(ChannelOption.TCP_NODELAY, true);
+ }
+
+ @Override
+ protected ByteBufOutputStream newPendingBuffer() {
+ return new ByteBufOutputStream(Unpooled.buffer());
+ }
+
+ @Override
+ protected void sendMessageChannel(Channel c, Object msg) {
+ c.writeAndFlush(msg);
+ }
+
+ @Override
+ protected void closeChannel(Channel c) {
+ c.close();
+ }
+
+ @Override
+ protected void resetPendingBuffer(ByteBufOutputStream b) {
+ b.buffer().clear();
+
+ }
+
+ @Override
+ protected void flushPendingBuffer(ByteBufOutputStream b, Channel c) {
+ c.writeAndFlush(b.buffer());
+ b.buffer().clear();
+
+ }
+
+ @Override
+ protected void closePendingBuffer(ByteBufOutputStream b) {
+ b.buffer().clear();
+
+ }
+
+ @Override
+ public void close() {
+ super.close();
+ group.shutdownGracefully();
+ }
+
+ @Override
+ protected void startConnection() {
+
+ try {
+ SocketAddress addr = (isSocks) ?
+ new IPAddress(System.getProperty("socksProxyHost", "localhost"), Integer.valueOf(System.getProperty("socksProxyPort", "0"))).getSocketAddress() :
+ session.getAddress().getSocketAddress();
+
+ ChannelFuture f = bootstrap.connect(addr);
+
+ f.addListener((ChannelFutureListener) future -> {
+
+ if (!future.isSuccess()) {
+ onConnectFailed(future.channel(), future.cause());
+ return;
+ }
+
+ Channel c = f.channel();
+
+ c.closeFuture().addListener((ChannelFutureListener) closeFt -> {
+ Channel ch = closeFt.channel();
+ onClosed(ch);
+ });
+
+ // MessageChannel mc = (MessageChannel) f.channel();
+
+ if (isSocks) {
+ // mc.setUseSocksProxy(isSocks);
+ LOG.debug("--- useSocksProxy ----");
+ LOG.debug(System.getProperty("socksProxyHost"));
+ LOG.debug(System.getProperty("socksProxyPort"));
+
+ }
+ });
+
+ } catch (NumberFormatException e) {
+ // TODO 自動生成された catch ブロック
+ e.printStackTrace();
+ } catch (UnknownHostException e) {
+ // TODO 自動生成された catch ブロック
+ e.printStackTrace();
+ }
+ }
+
+ public void onSocksConnected(Channel c) {
+ ChannelPipeline p = c.pipeline();
+ p.fireChannelActive();
+ }
-class NettyTcpClientTransport extends PooledStreamClientTransport {
- private static final String TCP_NO_DELAY = "tcpNoDelay";
-
- private final ClientBootstrap bootstrap;
-
- NettyTcpClientTransport(TcpClientConfig config, Session session,
- NettyEventLoop loop) {
- // TODO check session.getAddress() instanceof IPAddress
- super(config, session);
-
- RpcMessageHandler handler = new RpcMessageHandler(session);
-
- bootstrap = new ClientBootstrap(loop.getClientFactory());
- bootstrap.setPipelineFactory(new StreamPipelineFactory(loop.getMessagePack(), handler));
- Map options = config.getOptions();
- setIfNotPresent(options, TCP_NO_DELAY, Boolean.TRUE, bootstrap);
- bootstrap.setOptions(options);
- }
-
- private final ChannelFutureListener connectListener = new ChannelFutureListener() {
- public void operationComplete(ChannelFuture future) throws Exception {
- if (!future.isSuccess()) {
- onConnectFailed(future.getChannel(), future.getCause());
- return;
- }
- Channel c = future.getChannel();
- c.getCloseFuture().addListener(closeListener);
- onConnected(c);
- }
- };
-
- private final ChannelFutureListener closeListener = new ChannelFutureListener() {
- public void operationComplete(ChannelFuture future) throws Exception {
- Channel c = future.getChannel();
- onClosed(c);
- }
- };
-
- @Override
- protected void startConnection() {
- ChannelFuture f = bootstrap.connect(session.getAddress().getSocketAddress());
- f.addListener(connectListener);
- }
-
- @Override
- protected ChannelBufferOutputStream newPendingBuffer() {
- return new ChannelBufferOutputStream(
- ChannelBuffers.dynamicBuffer(HeapChannelBufferFactory.getInstance()));
- }
-
- @Override
- protected void resetPendingBuffer(ChannelBufferOutputStream b) {
- b.buffer().clear();
- }
-
- @Override
- protected void flushPendingBuffer(ChannelBufferOutputStream b, Channel c) {
- Channels.write(c, b.buffer());
- b.buffer().clear();
- }
-
- @Override
- protected void closePendingBuffer(ChannelBufferOutputStream b) {
- b.buffer().clear();
- }
-
- @Override
- protected void sendMessageChannel(Channel c, Object msg) {
- Channels.write(c, msg);
- }
-
- @Override
- protected void closeChannel(Channel c) {
- c.close();
- }
-
- private static void setIfNotPresent(Map options,
- String key, Object value, ClientBootstrap bootstrap) {
- if (!options.containsKey(key)) {
- bootstrap.setOption(key, value);
- }
- }
}
diff --git a/src/main/java/org/msgpack/rpc/loop/netty/NettyTcpServerTransport.java b/src/main/java/org/msgpack/rpc/loop/netty/NettyTcpServerTransport.java
index cab6cfe..9dcbd48 100644
--- a/src/main/java/org/msgpack/rpc/loop/netty/NettyTcpServerTransport.java
+++ b/src/main/java/org/msgpack/rpc/loop/netty/NettyTcpServerTransport.java
@@ -15,49 +15,86 @@
// See the License for the specific language governing permissions and
// limitations under the License.
//
+/*
+* Copyright (C) 2014-2015 Information Analysis Laboratory, NICT
+*
+* RaSC is free software: you can redistribute it and/or modify it
+* under the terms of the GNU Lesser General Public License as published by
+* the Free Software Foundation, either version 2.1 of the License, or (at
+* your option) any later version.
+*
+* RaSC is distributed in the hope that it will be useful, but
+* WITHOUT ANY WARRANTY; without even the implied warranty of
+* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser
+* General Public License for more details.
+*
+* You should have received a copy of the GNU Lesser General Public License
+* along with this program. If not, see options = config.getOptions();
- setIfNotPresent(options, CHILD_TCP_NODELAY, Boolean.TRUE, bootstrap);
- setIfNotPresent(options, REUSE_ADDRESS, Boolean.TRUE, bootstrap);
- bootstrap.setOptions(options);
- this.listenChannel = bootstrap.bind(address.getSocketAddress());
- }
-
- public void close() {
- listenChannel.close();
- }
-
- private static void setIfNotPresent(Map options,
- String key, Object value, ServerBootstrap bootstrap) {
- if (!options.containsKey(key)) {
- bootstrap.setOption(key, value);
- }
- }
+ private ChannelFuture future = null;
+ private final EventLoopGroup bossGroup = new NioEventLoopGroup(2);
+ private final EventLoopGroup workerGroup = new NioEventLoopGroup(8);
+
+ NettyTcpServerTransport(TcpServerConfig config, Server server, NettyEventLoop loop, Class rpcHandlerClass) {
+ if (server == null) {
+ throw new IllegalArgumentException("Server must not be null");
+ }
+
+ Address address = config.getListenAddress();
+
+ try {
+ RpcMessageHandler handler = rpcHandlerClass.getConstructor(Server.class).newInstance(server);
+ handler.useThread(true);
+
+ ServerBootstrap bootstrap = new ServerBootstrap().group(bossGroup, workerGroup).channel(NioServerSocketChannel.class);
+
+ bootstrap.childHandler(new ChannelInitializer() {
+
+ @Override
+ protected void initChannel(SocketChannel ch) throws Exception {
+ ChannelPipeline p = ch.pipeline();
+ p.addLast("msgpack-decode-stream", new MessagePackStreamDecoder(loop.getMessagePack()));
+ p.addLast("msgpack-encode", new MessagePackEncoder(loop.getMessagePack()));
+ p.addLast("message", new MessageHandler(handler));
+ }
+
+ });
+ bootstrap.childOption(ChannelOption.TCP_NODELAY, true);
+ bootstrap.option(ChannelOption.SO_REUSEADDR, true);
+ bootstrap.option(ChannelOption.TCP_NODELAY, true);
+ bootstrap.localAddress(address.getSocketAddress());
+ future = bootstrap.bind();
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+
+ // RpcMessageHandler handler = new RpcMessageHandlerEx(server);
+
+ }
+
+ public void close() {
+ future.channel().close();
+ workerGroup.shutdownGracefully();
+ bossGroup.shutdownGracefully();
+ }
}
diff --git a/src/main/java/org/msgpack/rpc/loop/netty/StreamPipelineFactory.java b/src/main/java/org/msgpack/rpc/loop/netty/StreamPipelineFactory.java
deleted file mode 100644
index 64a1cae..0000000
--- a/src/main/java/org/msgpack/rpc/loop/netty/StreamPipelineFactory.java
+++ /dev/null
@@ -1,42 +0,0 @@
-//
-// MessagePack-RPC for Java
-//
-// Copyright (C) 2010 FURUHASHI Sadayuki
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-//
-package org.msgpack.rpc.loop.netty;
-
-import org.jboss.netty.channel.Channels;
-import org.jboss.netty.channel.ChannelPipeline;
-import org.jboss.netty.channel.ChannelPipelineFactory;
-import org.msgpack.MessagePack;
-import org.msgpack.rpc.transport.RpcMessageHandler;
-
-class StreamPipelineFactory implements ChannelPipelineFactory {
- private RpcMessageHandler handler;
- private MessagePack messagePack;
-
- StreamPipelineFactory(MessagePack messagePack, RpcMessageHandler handler) {
- this.handler = handler;
- this.messagePack = messagePack;
- }
-
- public ChannelPipeline getPipeline() throws Exception {
- ChannelPipeline p = Channels.pipeline();
- p.addLast("msgpack-decode-stream", new MessagePackStreamDecoder(messagePack));
- p.addLast("msgpack-encode", new MessagePackEncoder(messagePack));
- p.addLast("message", new MessageHandler(handler));
- return p;
- }
-}
diff --git a/src/main/java/org/msgpack/rpc/reflect/ReflectionInvokerBuilder.java b/src/main/java/org/msgpack/rpc/reflect/ReflectionInvokerBuilder.java
index 2c403f2..cb153f5 100644
--- a/src/main/java/org/msgpack/rpc/reflect/ReflectionInvokerBuilder.java
+++ b/src/main/java/org/msgpack/rpc/reflect/ReflectionInvokerBuilder.java
@@ -15,6 +15,23 @@
// See the License for the specific language governing permissions and
// limitations under the License.
//
+/*
+* Copyright (C) 2014-2015 Information Analysis Laboratory, NICT
+*
+* RaSC is free software: you can redistribute it and/or modify it
+* under the terms of the GNU Lesser General Public License as published by
+* the Free Software Foundation, either version 2.1 of the License, or (at
+* your option) any later version.
+*
+* RaSC is distributed in the hope that it will be useful, but
+* WITHOUT ANY WARRANTY; without even the implied warranty of
+* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser
+* General Public License for more details.
+*
+* You should have received a copy of the GNU Lesser General Public License
+* along with this program. If not, see implements ClientTransport {
- private static final InternalLogger LOG =
- InternalLoggerFactory.getInstance(PooledStreamClientTransport.class);
-
- private final Object lock = new Object();
- private final List pool = new ArrayList();
- private final List errorChannelPool = new ArrayList();
- private int reconnectionLimit;
- private int connecting = 0;
-
- protected final Session session;
- protected final StreamClientConfig config;
- protected final MessagePack messagePack;
-
- public PooledStreamClientTransport(StreamClientConfig config,
- Session session) {
- this.session = session;
- this.config = config;
- this.reconnectionLimit = config.getReconnectionLimit();
- this.messagePack = session.getEventLoop().getMessagePack();
- }
-
- protected Session getSession() {
- return session;
- }
-
- protected StreamClientConfig getConfig() {
- return config;
- }
-
- public void sendMessage(Object msg) {
- synchronized (lock) {
- if (connecting == -1) {
- return;
- } // already closed
- if (pool.isEmpty()) {
- if (connecting == 0) {
- connecting++;
- startConnection();
- }
- if (pool.isEmpty()) { // may be already connected
- try {
- messagePack.write(getPendingBuffer(), msg);
- } catch (IOException e) {
- // FIXME
- }
- return;
- }
- }
- // FIXME pseudo connection load balancing
- Channel c = pool.get(0);
- sendMessageChannel(c, msg);
- }
- }
-
- public void close() {
- synchronized (lock) {
- LOG.info("Close all channels");
- if (pendingBuffer != null) {
- closePendingBuffer(pendingBuffer);
- pendingBuffer = null;
- }
- connecting = -1;
- for (Channel c : pool) {
- closeChannel(c);
- }
- for (Channel c : errorChannelPool) {
- closeChannel(c);
- }
- pool.clear();
- errorChannelPool.clear();
- }
- }
-
- public void onConnected(Channel c) {
- synchronized (lock) {
- if (connecting == -1) {
- closeChannel(c);
- return;
- } // already closed
- LOG.debug("Success to connect new channel " + c);
- pool.add(c);
- connecting = 0;
- if (pendingBuffer != null) {
- flushPendingBuffer(pendingBuffer, c);
- }
- }
- }
-
- public void onConnectFailed(Channel c, Throwable cause) {
- synchronized (lock) {
- if (connecting == -1) {
- return;
- } // already closed
- if (connecting < reconnectionLimit) {
- LOG.info(String.format("Reconnect %s(retry:%s)", c,
- connecting + 1), cause);
- connecting++;
- if (pool.remove(c)) {// remove error channel
- errorChannelPool.add(c);
- }
- startConnection();
- } else {
- LOG.error(String.format(
- "Fail to connect %s(tried %s times)", c,
- reconnectionLimit), cause);
- connecting = 0;
- if (pendingBuffer != null) {
- resetPendingBuffer(pendingBuffer);
- }
- session.transportConnectFailed();
- }
- }
- }
-
- public void onClosed(Channel c) {
- synchronized (lock) {
- if (connecting == -1) {
- return;
- } // already closed
- LOG.info(String.format("Close channel %s", c));
- pool.remove(c);
- errorChannelPool.remove(c);
- }
- }
-
- private PendingBuffer pendingBuffer = null;
-
- protected PendingBuffer getPendingBuffer() {
- if (pendingBuffer == null) {
- pendingBuffer = newPendingBuffer();
- }
- return pendingBuffer;
- }
-
- protected abstract PendingBuffer newPendingBuffer();
-
- protected abstract void resetPendingBuffer(PendingBuffer b);
-
- protected abstract void flushPendingBuffer(PendingBuffer b, Channel c);
-
- protected abstract void closePendingBuffer(PendingBuffer b);
-
- protected abstract void startConnection();
-
- protected abstract void sendMessageChannel(Channel c, Object msg);
-
- protected abstract void closeChannel(Channel c);
+ private static final InternalLogger LOG = InternalLoggerFactory.getInstance(PooledStreamClientTransport.class);
+ private static final int MAX_CHANNEL = 2;
+
+ private final List pool = new CopyOnWriteArrayList<>();
+ private final List errorChannelPool = new CopyOnWriteArrayList<>();
+ private final AtomicInteger connecting = new AtomicInteger();
+ private final Queue queMessage = new ConcurrentLinkedQueue<>();
+ private final AtomicInteger assignIndex = new AtomicInteger();
+ protected final Session session;
+ protected final StreamClientConfig config;
+ protected final MessagePack messagePack;
+ protected final int maxChannel = Objects.isNull(System.getProperty("maxChannel")) ? MAX_CHANNEL : Integer.valueOf(System.getProperty("maxChannel"));
+ protected final boolean isSocks = (Objects.nonNull(System.getProperty("socksProxyHost")) && Objects.nonNull(System.getProperty("socksProxyPort")));
+
+ public PooledStreamClientTransport(StreamClientConfig config, Session session) {
+ this.session = session;
+ this.config = config;
+ this.messagePack = session.getEventLoop().getMessagePack();
+ }
+
+ protected Session getSession() {
+ return session;
+ }
+
+ protected StreamClientConfig getConfig() {
+ return config;
+ }
+
+ public synchronized void sendMessage(Object msg) {
+ if (connecting.get() == -1) {
+ return;
+ } // already closed
+
+ if (connecting.get() <= 0) {
+ if (connecting.get() < maxChannel) {
+ connecting.incrementAndGet();
+ startConnection();
+ queMessage.offer(msg);
+ } else {
+ queMessage.offer(msg);
+ }
+ } else {
+ if (connecting.get() >= maxChannel) {
+ if (!pool.isEmpty()) {
+ int index = assignIndex.getAndIncrement() % pool.size();
+ if (index >= Integer.MAX_VALUE) {
+ assignIndex.set(0);
+ }
+ Channel c = pool.get(index);
+ sendMessageChannel(c, msg);
+ flushMessageQue(c);
+ } else {
+ queMessage.offer(msg);
+ }
+ } else {
+ connecting.incrementAndGet();
+ startConnection();
+ queMessage.offer(msg);
+ }
+ }
+ }
+
+ public void close() {
+ LOG.debug("Close all channels");
+ connecting.set(-1);
+ for (Channel c : pool) {
+ closeChannel(c);
+ }
+ for (Channel c : errorChannelPool) {
+ closeChannel(c);
+ }
+ pool.clear();
+ errorChannelPool.clear();
+ queMessage.clear();
+ }
+
+ public void onConnected(Channel c) {
+ if (connecting.get() == -1) {
+ closeChannel(c);
+ return;
+ } // already closed
+
+ LOG.debug("Success to connect new channel " + c);
+ LOG.debug(Objects.toString(c));
+ pool.add(c);
+ flushMessageQue(c);
+ }
+
+ public void onConnectFailed(Channel c, Throwable cause) {
+ if (connecting.get() == -1) {
+ return;
+ } // already closed
+ LOG.error(String.format("Fail to connect %s", c), cause);
+ connecting.set(pool.size());
+ session.transportConnectFailed();
+ closeChannel(c);
+ }
+
+ public void onClosed(Channel c) {
+ if (connecting.get() == -1) {
+ return;
+ } // already closed
+ LOG.debug(String.format("Close channel %s", c));
+ pool.remove(c);
+ errorChannelPool.remove(c);
+ connecting.set(pool.size());
+ }
+
+ @Override
+ public void sendDataDelay(Object obj) {
+
+ }
+
+ public void onError(Channel c, String msg) {
+ if (connecting.get() == -1) {
+ return;
+ } // already closed
+ LOG.info(String.format("Error channel %s", c));
+ closeChannel(c);
+ session.transportError(msg);
+
+ }
+
+ protected synchronized void flushMessageQue(Channel c) {
+ if (!queMessage.isEmpty()) {
+ queMessage.forEach(m -> sendMessageChannel(c, m));
+ queMessage.clear();
+ }
+ }
+
+ protected abstract PendingBuffer newPendingBuffer();
+
+ protected abstract void resetPendingBuffer(PendingBuffer b);
+
+ protected abstract void flushPendingBuffer(PendingBuffer b, Channel c);
+
+ protected abstract void closePendingBuffer(PendingBuffer b);
+
+ protected abstract void sendMessageChannel(Channel c, Object msg);
+
+ protected abstract void closeChannel(Channel c);
+
+ protected abstract void startConnection();
+
}
diff --git a/src/test/java/org/msgpack/rpc/reflect/AnnotationTest.java b/src/test/java/org/msgpack/rpc/reflect/AnnotationTest.java
index 3a8665f..a9cd4d9 100644
--- a/src/test/java/org/msgpack/rpc/reflect/AnnotationTest.java
+++ b/src/test/java/org/msgpack/rpc/reflect/AnnotationTest.java
@@ -15,18 +15,29 @@
// See the License for the specific language governing permissions and
// limitations under the License.
//
+/*
+* Copyright (C) 2014-2015 Information Analysis Laboratory, NICT
+*
+* RaSC is free software: you can redistribute it and/or modify it
+* under the terms of the GNU Lesser General Public License as published by
+* the Free Software Foundation, either version 2.1 of the License, or (at
+* your option) any later version.
+*
+* RaSC is distributed in the hope that it will be useful, but
+* WITHOUT ANY WARRANTY; without even the implied warranty of
+* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser
+* General Public License for more details.
+*
+* You should have received a copy of the GNU Lesser General Public License
+* along with this program. If not, see