You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by ni...@apache.org on 2015/02/11 14:28:54 UTC
[1/3] camel git commit: CAMEL-1077 Added remote address support to
camel-netty4
Repository: camel
Updated Branches:
refs/heads/master d1b1db91f -> aaa6f5582
CAMEL-1077 Added remote address support to camel-netty4
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/0b685e8e
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/0b685e8e
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/0b685e8e
Branch: refs/heads/master
Commit: 0b685e8e2feaf1e020cca3b1bd255c6284abc43e
Parents: d1b1db9
Author: Willem Jiang <wi...@gmail.com>
Authored: Wed Feb 11 21:14:19 2015 +0800
Committer: Willem Jiang <wi...@gmail.com>
Committed: Wed Feb 11 21:14:19 2015 +0800
----------------------------------------------------------------------
...lientModeTCPNettyServerBootstrapFactory.java | 214 +++++++++++++++++++
.../component/netty4/NettyConfiguration.java | 10 +
.../camel/component/netty4/NettyConsumer.java | 6 +-
.../netty4/NettyConsumerClientModeTest.java | 160 ++++++++++++++
4 files changed, 389 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/0b685e8e/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/ClientModeTCPNettyServerBootstrapFactory.java
----------------------------------------------------------------------
diff --git a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/ClientModeTCPNettyServerBootstrapFactory.java b/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/ClientModeTCPNettyServerBootstrapFactory.java
new file mode 100644
index 0000000..ac734e3
--- /dev/null
+++ b/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/ClientModeTCPNettyServerBootstrapFactory.java
@@ -0,0 +1,214 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.apache.camel.component.netty4;
+
+import java.net.ConnectException;
+import java.net.InetSocketAddress;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelOption;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.group.ChannelGroup;
+import io.netty.channel.group.DefaultChannelGroup;
+import io.netty.channel.socket.nio.NioSocketChannel;
+import io.netty.util.concurrent.ImmediateEventExecutor;
+import org.apache.camel.CamelContext;
+import org.apache.camel.CamelException;
+import org.apache.camel.support.ServiceSupport;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A {@link NettyServerBootstrapFactory} which is used by a single consumer (not shared).
+ */
+public class ClientModeTCPNettyServerBootstrapFactory extends ServiceSupport implements NettyServerBootstrapFactory {
+
+ protected static final Logger LOG = LoggerFactory.getLogger(ClientModeTCPNettyServerBootstrapFactory.class);
+ private final ChannelGroup allChannels;
+ private CamelContext camelContext;
+ private ThreadFactory threadFactory;
+ private NettyServerBootstrapConfiguration configuration;
+ private ChannelInitializer<Channel> pipelineFactory;
+ private Bootstrap clientBootstrap;
+ private Channel channel;
+ private EventLoopGroup workerGroup;
+
+ public ClientModeTCPNettyServerBootstrapFactory() {
+ // The executor just execute tasks in the callers thread
+ this.allChannels = new DefaultChannelGroup(ClientModeTCPNettyServerBootstrapFactory.class.getName(), ImmediateEventExecutor.INSTANCE);
+ }
+
+ public void init(CamelContext camelContext, NettyServerBootstrapConfiguration configuration, ChannelInitializer<Channel> pipelineFactory) {
+ this.camelContext = camelContext;
+ this.configuration = configuration;
+ this.pipelineFactory = pipelineFactory;
+ }
+
+ public void init(ThreadFactory threadFactory, NettyServerBootstrapConfiguration configuration, ChannelInitializer<Channel> pipelineFactory) {
+ this.threadFactory = threadFactory;
+ this.configuration = configuration;
+ this.pipelineFactory = pipelineFactory;
+ }
+
+ public void addChannel(Channel channel) {
+ allChannels.add(channel);
+ }
+
+ public void removeChannel(Channel channel) {
+ allChannels.remove(channel);
+ }
+
+ public void addConsumer(NettyConsumer consumer) {
+ // does not allow sharing
+ }
+
+ public void removeConsumer(NettyConsumer consumer) {
+ // does not allow sharing
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ if (camelContext == null && threadFactory == null) {
+ throw new IllegalArgumentException("Either CamelContext or ThreadFactory must be set on " + this);
+ }
+ startServerBootstrap();
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ stopServerBootstrap();
+ }
+
+ @Override
+ protected void doResume() throws Exception {
+ LOG.debug("ClientModeServerBootstrap connect to {}:{}", configuration.getHost(), configuration.getPort());
+ ChannelFuture connectFuture = clientBootstrap.connect(new InetSocketAddress(configuration.getHost(), configuration.getPort()));
+ channel = openChannel(connectFuture);
+ // to keep track of all channels in use
+ allChannels.add(channel);
+ }
+
+ @Override
+ protected void doSuspend() throws Exception {
+ if (channel != null) {
+ LOG.debug("ClientModeServerBootstrap unbinding from {}:{}", configuration.getHost(), configuration.getPort());
+ ChannelFuture future = channel.close();
+ future.awaitUninterruptibly();
+ allChannels.remove(channel);
+ channel = null;
+ }
+ }
+
+ protected void startServerBootstrap() throws Exception {
+ // prefer using explicit configured thread pools
+
+ EventLoopGroup wg = configuration.getWorkerGroup();
+
+ if (wg == null) {
+ // create new pool which we should shutdown when stopping as its not shared
+ workerGroup = new NettyWorkerPoolBuilder()
+ .withWorkerCount(configuration.getWorkerCount())
+ .withName("NettyServerTCPWorker")
+ .build();
+ wg = workerGroup;
+ }
+
+ clientBootstrap = new Bootstrap();
+ clientBootstrap.channel(NioSocketChannel.class);
+ clientBootstrap.group(wg);
+ clientBootstrap.option(ChannelOption.SO_KEEPALIVE, configuration.isKeepAlive());
+ clientBootstrap.option(ChannelOption.TCP_NODELAY, configuration.isTcpNoDelay());
+ clientBootstrap.option(ChannelOption.SO_REUSEADDR, configuration.isReuseAddress());
+ clientBootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, configuration.getConnectTimeout());
+
+
+ LOG.debug("Created ClientBootstrap {}", clientBootstrap);
+ clientBootstrap.handler(pipelineFactory);
+ ChannelFuture channelFuture = clientBootstrap.connect(new InetSocketAddress(configuration.getHost(), configuration.getPort()));
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Created new TCP client bootstrap connecting to {}:{} with options: {}",
+ new Object[]{configuration.getHost(), configuration.getPort(), clientBootstrap});
+ }
+ LOG.info("ClientModeServerBootstrap binding to {}:{}", configuration.getHost(), configuration.getPort());
+ channel = openChannel(channelFuture);
+ // to keep track of all channels in use
+ allChannels.add(channel);
+ }
+
+ protected void stopServerBootstrap() {
+ // close all channels
+ LOG.info("ClientModeServerBootstrap unbinding from {}:{}", configuration.getHost(), configuration.getPort());
+
+ LOG.trace("Closing {} channels", allChannels.size());
+ if (allChannels != null) {
+ allChannels.close().awaitUninterruptibly();
+ }
+
+ if (workerGroup != null) {
+ workerGroup.shutdownGracefully();
+ workerGroup = null;
+ }
+ }
+
+ protected Channel openChannel(ChannelFuture channelFuture) throws Exception {
+ // blocking for channel to be done
+ if (LOG.isTraceEnabled()) {
+ LOG.trace("Waiting for operation to complete {} for {} millis", channelFuture, configuration.getConnectTimeout());
+ }
+ // here we need to wait it in other thread
+ final CountDownLatch channelLatch = new CountDownLatch(1);
+ channelFuture.addListener(new ChannelFutureListener() {
+ @Override
+ public void operationComplete(ChannelFuture cf) throws Exception {
+ channelLatch.countDown();
+ }
+ });
+
+ try {
+ channelLatch.await(configuration.getConnectTimeout(), TimeUnit.MILLISECONDS);
+ } catch (InterruptedException ex) {
+ throw new CamelException("Interrupted while waiting for " + "connection to "
+ + configuration.getAddress());
+ }
+
+
+ if (!channelFuture.isDone() || !channelFuture.isSuccess()) {
+ ConnectException cause = new ConnectException("Cannot connect to " + configuration.getAddress());
+ if (channelFuture.cause() != null) {
+ cause.initCause(channelFuture.cause());
+ }
+ throw cause;
+ }
+ Channel answer = channelFuture.channel();
+
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Creating connector to address: {}", configuration.getAddress());
+ }
+
+ return answer;
+
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/camel/blob/0b685e8e/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyConfiguration.java
----------------------------------------------------------------------
diff --git a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyConfiguration.java b/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyConfiguration.java
index 467545d..2567d42 100644
--- a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyConfiguration.java
+++ b/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyConfiguration.java
@@ -92,6 +92,8 @@ public class NettyConfiguration extends NettyServerBootstrapConfiguration implem
private boolean producerPoolEnabled = true;
@UriParam(defaultValue = "false")
private boolean udpConnectionlessSending;
+ @UriParam(defaultValue = "false")
+ private boolean clientMode;
/**
* Returns a copy of this configuration
@@ -481,6 +483,14 @@ public class NettyConfiguration extends NettyServerBootstrapConfiguration implem
public void setUdpConnectionlessSending(boolean udpConnectionlessSending) {
this.udpConnectionlessSending = udpConnectionlessSending;
}
+
+ public boolean isClientMode() {
+ return clientMode;
+ }
+
+ public void setClientMode(boolean clientMode) {
+ this.clientMode = clientMode;
+ }
private static <T> void addToHandlersList(List<T> configured, List<T> handlers, Class<T> handlerType) {
if (handlers != null) {
http://git-wip-us.apache.org/repos/asf/camel/blob/0b685e8e/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyConsumer.java b/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyConsumer.java
index f145a75..574e9dc 100644
--- a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyConsumer.java
+++ b/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyConsumer.java
@@ -59,7 +59,11 @@ public class NettyConsumer extends DefaultConsumer {
}
if (isTcp()) {
- nettyServerBootstrapFactory = new SingleTCPNettyServerBootstrapFactory();
+ if (configuration.isClientMode()) {
+ nettyServerBootstrapFactory = new ClientModeTCPNettyServerBootstrapFactory();
+ } else {
+ nettyServerBootstrapFactory = new SingleTCPNettyServerBootstrapFactory();
+ }
} else {
nettyServerBootstrapFactory = new SingleUDPNettyServerBootstrapFactory();
}
http://git-wip-us.apache.org/repos/asf/camel/blob/0b685e8e/components/camel-netty4/src/test/java/org/apache/camel/component/netty4/NettyConsumerClientModeTest.java
----------------------------------------------------------------------
diff --git a/components/camel-netty4/src/test/java/org/apache/camel/component/netty4/NettyConsumerClientModeTest.java b/components/camel-netty4/src/test/java/org/apache/camel/component/netty4/NettyConsumerClientModeTest.java
new file mode 100644
index 0000000..88a219c
--- /dev/null
+++ b/components/camel-netty4/src/test/java/org/apache/camel/component/netty4/NettyConsumerClientModeTest.java
@@ -0,0 +1,160 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.apache.camel.component.netty4;
+
+
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioServerSocketChannel;
+import io.netty.handler.codec.DelimiterBasedFrameDecoder;
+import io.netty.handler.codec.Delimiters;
+import io.netty.handler.codec.string.StringDecoder;
+import io.netty.handler.codec.string.StringEncoder;
+import org.apache.camel.Exchange;
+import org.apache.camel.Processor;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.junit.Test;
+
+public class NettyConsumerClientModeTest extends BaseNettyTest {
+ private MyServer server;
+
+
+ public void startNettyServer() throws Exception {
+ server = new MyServer(getPort());
+ server.start();
+ }
+
+ public void shutdownServer() {
+ if (server != null) {
+ server.shutdown();
+ }
+ }
+ @Test
+ public void testNettyRoute() throws Exception {
+ try {
+ startNettyServer();
+ MockEndpoint receive = context.getEndpoint("mock:receive", MockEndpoint.class);
+ receive.expectedBodiesReceived("Bye Willem");
+ context.startRoute("client");
+ receive.assertIsSatisfied();
+ } finally {
+ shutdownServer();
+ }
+
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ @Override
+ public void configure() throws Exception {
+ from("netty4:tcp://localhost:{{port}}?textline=true&clientMode=true").id("client")
+ .process(new Processor() {
+ public void process(final Exchange exchange) {
+ String body = exchange.getIn().getBody(String.class);
+ exchange.getOut().setBody("Bye " + body);
+ }
+ }).to("mock:receive").noAutoStartup();
+ }
+ };
+ }
+
+ private static class MyServer {
+ private int port;
+ private ServerBootstrap bootstrap;
+ private Channel channel;
+ private EventLoopGroup bossGroup;
+ private EventLoopGroup workerGroup;
+
+ public MyServer(int port) {
+ this.port = port;
+ }
+
+ public void start() throws Exception {
+ bossGroup = new NioEventLoopGroup(1);
+ workerGroup = new NioEventLoopGroup();
+
+ bootstrap = new ServerBootstrap();
+ bootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class)
+ .childHandler(new ServerInitializer());
+
+ ChannelFuture cf = bootstrap.bind(port).sync();
+ channel = cf.channel();
+
+ }
+
+ public void shutdown() {
+ channel.close();
+ bossGroup.shutdownGracefully();
+ workerGroup.shutdownGracefully();
+ }
+
+ }
+
+ private static class ServerHandler extends SimpleChannelInboundHandler<String> {
+
+ public void channelActive(ChannelHandlerContext ctx) throws Exception {
+ ctx.write("Willem\r\n");
+ ctx.flush();
+ }
+
+ @Override
+ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
+ cause.printStackTrace();
+ ctx.close();
+ }
+ @Override
+ protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
+ // Do nothing here
+ }
+
+ public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
+ ctx.flush();
+ }
+ }
+
+ private static class ServerInitializer extends ChannelInitializer<SocketChannel> {
+ private static final StringDecoder DECODER = new StringDecoder();
+ private static final StringEncoder ENCODER = new StringEncoder();
+ private static final ServerHandler SERVERHANDLER = new ServerHandler();
+
+ @Override
+ public void initChannel(SocketChannel ch) throws Exception {
+ ChannelPipeline pipeline = ch.pipeline();
+
+ // Add the text line codec combination first,
+ pipeline.addLast("framer", new DelimiterBasedFrameDecoder(
+ 8192, Delimiters.lineDelimiter()));
+ // the encoder and decoder are static as these are sharable
+ pipeline.addLast("decoder", DECODER);
+ pipeline.addLast("encoder", ENCODER);
+
+ // and then business logic.
+ pipeline.addLast("handler", SERVERHANDLER);
+ }
+ }
+
+}
[2/3] camel git commit: CAMEL-1077 Remove the allChannel as we don't
need to track the channel in client mode
Posted by ni...@apache.org.
CAMEL-1077 Remove the allChannel as we don't need to track the channel in client mode
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/4e8f6c01
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/4e8f6c01
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/4e8f6c01
Branch: refs/heads/master
Commit: 4e8f6c01f020c3161467f312c21fce3beda74993
Parents: 0b685e8
Author: Willem Jiang <wi...@gmail.com>
Authored: Wed Feb 11 21:19:33 2015 +0800
Committer: Willem Jiang <wi...@gmail.com>
Committed: Wed Feb 11 21:19:33 2015 +0800
----------------------------------------------------------------------
...lientModeTCPNettyServerBootstrapFactory.java | 22 +++++---------------
1 file changed, 5 insertions(+), 17 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/4e8f6c01/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/ClientModeTCPNettyServerBootstrapFactory.java
----------------------------------------------------------------------
diff --git a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/ClientModeTCPNettyServerBootstrapFactory.java b/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/ClientModeTCPNettyServerBootstrapFactory.java
index ac734e3..38c6c65 100644
--- a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/ClientModeTCPNettyServerBootstrapFactory.java
+++ b/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/ClientModeTCPNettyServerBootstrapFactory.java
@@ -45,7 +45,7 @@ import org.slf4j.LoggerFactory;
public class ClientModeTCPNettyServerBootstrapFactory extends ServiceSupport implements NettyServerBootstrapFactory {
protected static final Logger LOG = LoggerFactory.getLogger(ClientModeTCPNettyServerBootstrapFactory.class);
- private final ChannelGroup allChannels;
+
private CamelContext camelContext;
private ThreadFactory threadFactory;
private NettyServerBootstrapConfiguration configuration;
@@ -55,8 +55,6 @@ public class ClientModeTCPNettyServerBootstrapFactory extends ServiceSupport imp
private EventLoopGroup workerGroup;
public ClientModeTCPNettyServerBootstrapFactory() {
- // The executor just execute tasks in the callers thread
- this.allChannels = new DefaultChannelGroup(ClientModeTCPNettyServerBootstrapFactory.class.getName(), ImmediateEventExecutor.INSTANCE);
}
public void init(CamelContext camelContext, NettyServerBootstrapConfiguration configuration, ChannelInitializer<Channel> pipelineFactory) {
@@ -72,11 +70,11 @@ public class ClientModeTCPNettyServerBootstrapFactory extends ServiceSupport imp
}
public void addChannel(Channel channel) {
- allChannels.add(channel);
+ // we don't need to track the channel in client mode
}
public void removeChannel(Channel channel) {
- allChannels.remove(channel);
+ // we don't need to track the channel in client mode
}
public void addConsumer(NettyConsumer consumer) {
@@ -105,17 +103,13 @@ public class ClientModeTCPNettyServerBootstrapFactory extends ServiceSupport imp
LOG.debug("ClientModeServerBootstrap connect to {}:{}", configuration.getHost(), configuration.getPort());
ChannelFuture connectFuture = clientBootstrap.connect(new InetSocketAddress(configuration.getHost(), configuration.getPort()));
channel = openChannel(connectFuture);
- // to keep track of all channels in use
- allChannels.add(channel);
}
@Override
protected void doSuspend() throws Exception {
if (channel != null) {
LOG.debug("ClientModeServerBootstrap unbinding from {}:{}", configuration.getHost(), configuration.getPort());
- ChannelFuture future = channel.close();
- future.awaitUninterruptibly();
- allChannels.remove(channel);
+ channel.close().sync();
channel = null;
}
}
@@ -152,19 +146,13 @@ public class ClientModeTCPNettyServerBootstrapFactory extends ServiceSupport imp
}
LOG.info("ClientModeServerBootstrap binding to {}:{}", configuration.getHost(), configuration.getPort());
channel = openChannel(channelFuture);
- // to keep track of all channels in use
- allChannels.add(channel);
+
}
protected void stopServerBootstrap() {
// close all channels
LOG.info("ClientModeServerBootstrap unbinding from {}:{}", configuration.getHost(), configuration.getPort());
- LOG.trace("Closing {} channels", allChannels.size());
- if (allChannels != null) {
- allChannels.close().awaitUninterruptibly();
- }
-
if (workerGroup != null) {
workerGroup.shutdownGracefully();
workerGroup = null;
[3/3] camel git commit: CAMEL-1077 Remove the allChannels as we don't
need to track the channel in client mode
Posted by ni...@apache.org.
CAMEL-1077 Remove the allChannels as we don't need to track the channel in client mode
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/aaa6f558
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/aaa6f558
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/aaa6f558
Branch: refs/heads/master
Commit: aaa6f5582fe9fb983f2468bfd8247e5e85de601c
Parents: 4e8f6c0
Author: Willem Jiang <wi...@gmail.com>
Authored: Wed Feb 11 21:25:30 2015 +0800
Committer: Willem Jiang <wi...@gmail.com>
Committed: Wed Feb 11 21:25:30 2015 +0800
----------------------------------------------------------------------
...lientModeTCPNettyServerBootstrapFactory.java | 37 ++++----------------
1 file changed, 7 insertions(+), 30 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/aaa6f558/components/camel-netty/src/main/java/org/apache/camel/component/netty/ClientModeTCPNettyServerBootstrapFactory.java
----------------------------------------------------------------------
diff --git a/components/camel-netty/src/main/java/org/apache/camel/component/netty/ClientModeTCPNettyServerBootstrapFactory.java b/components/camel-netty/src/main/java/org/apache/camel/component/netty/ClientModeTCPNettyServerBootstrapFactory.java
index 1b06127..0717f4b 100644
--- a/components/camel-netty/src/main/java/org/apache/camel/component/netty/ClientModeTCPNettyServerBootstrapFactory.java
+++ b/components/camel-netty/src/main/java/org/apache/camel/component/netty/ClientModeTCPNettyServerBootstrapFactory.java
@@ -32,9 +32,6 @@ import org.jboss.netty.channel.ChannelFactory;
import org.jboss.netty.channel.ChannelFuture;
import org.jboss.netty.channel.ChannelFutureListener;
import org.jboss.netty.channel.ChannelPipelineFactory;
-import org.jboss.netty.channel.group.ChannelGroup;
-import org.jboss.netty.channel.group.ChannelGroupFuture;
-import org.jboss.netty.channel.group.DefaultChannelGroup;
import org.jboss.netty.channel.socket.nio.BossPool;
import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;
import org.jboss.netty.channel.socket.nio.WorkerPool;
@@ -48,7 +45,6 @@ import org.slf4j.LoggerFactory;
public class ClientModeTCPNettyServerBootstrapFactory extends ServiceSupport implements NettyServerBootstrapFactory {
protected static final Logger LOG = LoggerFactory.getLogger(ClientModeTCPNettyServerBootstrapFactory.class);
- private final ChannelGroup allChannels;
private CamelContext camelContext;
private ThreadFactory threadFactory;
private NettyServerBootstrapConfiguration configuration;
@@ -58,10 +54,7 @@ public class ClientModeTCPNettyServerBootstrapFactory extends ServiceSupport imp
private Channel channel;
private BossPool bossPool;
private WorkerPool workerPool;
-
- public ClientModeTCPNettyServerBootstrapFactory() {
- this.allChannels = new DefaultChannelGroup(ClientModeTCPNettyServerBootstrapFactory.class.getName());
- }
+
public void init(CamelContext camelContext, NettyServerBootstrapConfiguration configuration, ChannelPipelineFactory pipelineFactory) {
this.camelContext = camelContext;
@@ -76,11 +69,11 @@ public class ClientModeTCPNettyServerBootstrapFactory extends ServiceSupport imp
}
public void addChannel(Channel channel) {
- allChannels.add(channel);
+ // we don't need to track the channel in client mode
}
public void removeChannel(Channel channel) {
- allChannels.remove(channel);
+ // we don't need to track the channel in client mode
}
public void addConsumer(NettyConsumer consumer) {
@@ -106,26 +99,16 @@ public class ClientModeTCPNettyServerBootstrapFactory extends ServiceSupport imp
@Override
protected void doResume() throws Exception {
- if (channel != null) {
- LOG.debug("ServerBootstrap connecting to {}:{}", configuration.getHost(), configuration.getPort());
- ChannelFuture future = channel.connect(new InetSocketAddress(configuration.getHost(), configuration.getPort()));
- future.awaitUninterruptibly();
- if (!future.isSuccess()) {
- // if we cannot connect, then re-create channel
- allChannels.remove(channel);
- ChannelFuture connectFuture = serverBootstrap.connect(new InetSocketAddress(configuration.getHost(), configuration.getPort()));
- channel = openChannel(connectFuture);
- allChannels.add(channel);
- }
- }
+ ChannelFuture connectFuture = serverBootstrap.connect(new InetSocketAddress(configuration.getHost(), configuration.getPort()));
+ channel = openChannel(connectFuture);
}
@Override
protected void doSuspend() throws Exception {
if (channel != null) {
LOG.debug("ServerBootstrap disconnecting from {}:{}", configuration.getHost(), configuration.getPort());
- ChannelFuture future = channel.disconnect();
- future.awaitUninterruptibly();
+ channel.disconnect().sync();
+ channel = null;
}
}
@@ -213,8 +196,6 @@ public class ClientModeTCPNettyServerBootstrapFactory extends ServiceSupport imp
throw cause;
}
Channel answer = channelFuture.getChannel();
- // to keep track of all channels in use
- allChannels.add(answer);
if (LOG.isDebugEnabled()) {
LOG.debug("Creating connector to address: {}", configuration.getAddress());
@@ -226,10 +207,6 @@ public class ClientModeTCPNettyServerBootstrapFactory extends ServiceSupport imp
// close all channels
LOG.info("ServerBootstrap disconnecting from {}:{}", configuration.getHost(), configuration.getPort());
- LOG.trace("Closing {} channels", allChannels.size());
- ChannelGroupFuture future = allChannels.close();
- future.awaitUninterruptibly();
-
// close server external resources
if (channelFactory != null) {
channelFactory.releaseExternalResources();