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 2014/07/18 17:15:37 UTC
[10/12] git commit: CAMEL-6555 Try to fix the compile errors in
camel-netty4 src
CAMEL-6555 Try to fix the compile errors in camel-netty4 src
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/7c89323a
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/7c89323a
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/7c89323a
Branch: refs/heads/CAMEL-6555
Commit: 7c89323adde43fa26443830737c4bacd1abc84b0
Parents: 2e85d5a
Author: Willem Jiang <wi...@gmail.com>
Authored: Fri Jul 18 15:52:22 2014 +0800
Committer: Willem Jiang <wi...@gmail.com>
Committed: Fri Jul 18 16:45:07 2014 +0800
----------------------------------------------------------------------
.../netty4/DefaultClientPipelineFactory.java | 2 +-
.../netty4/DefaultServerPipelineFactory.java | 24 ++---
.../camel/component/netty4/NettyComponent.java | 23 +++--
.../component/netty4/NettyConfiguration.java | 3 +-
.../camel/component/netty4/NettyEndpoint.java | 4 +-
.../camel/component/netty4/NettyProducer.java | 44 +++++----
.../NettyServerBootstrapConfiguration.java | 40 ++++----
.../netty4/NettyServerBootstrapFactory.java | 14 +--
.../netty4/NettyServerBossPoolBuilder.java | 12 +--
.../netty4/NettyWorkerPoolBuilder.java | 17 ++--
.../SingleTCPNettyServerBootstrapFactory.java | 96 ++++++++++----------
.../SingleUDPNettyServerBootstrapFactory.java | 25 +++--
12 files changed, 161 insertions(+), 143 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/7c89323a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/DefaultClientPipelineFactory.java
----------------------------------------------------------------------
diff --git a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/DefaultClientPipelineFactory.java b/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/DefaultClientPipelineFactory.java
index 0c0f4d6..3e5998a 100644
--- a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/DefaultClientPipelineFactory.java
+++ b/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/DefaultClientPipelineFactory.java
@@ -16,7 +16,6 @@
*/
package org.apache.camel.component.netty4;
-import java.nio.channels.Channels;
import java.util.List;
import java.util.concurrent.TimeUnit;
@@ -28,6 +27,7 @@ import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelPipeline;
import io.netty.handler.ssl.SslHandler;
import io.netty.handler.timeout.ReadTimeoutHandler;
+
import org.apache.camel.component.netty4.handlers.ClientChannelHandler;
import org.apache.camel.component.netty4.ssl.SSLEngineFactory;
import org.apache.camel.util.ObjectHelper;
http://git-wip-us.apache.org/repos/asf/camel/blob/7c89323a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/DefaultServerPipelineFactory.java
----------------------------------------------------------------------
diff --git a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/DefaultServerPipelineFactory.java b/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/DefaultServerPipelineFactory.java
index f981fce..cf9e57e 100644
--- a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/DefaultServerPipelineFactory.java
+++ b/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/DefaultServerPipelineFactory.java
@@ -28,6 +28,7 @@ import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelPipeline;
import io.netty.handler.ssl.SslHandler;
+import io.netty.util.concurrent.EventExecutorGroup;
import org.apache.camel.component.netty4.handlers.ServerChannelHandler;
import org.apache.camel.component.netty4.ssl.SSLEngineFactory;
import org.apache.camel.util.ObjectHelper;
@@ -101,18 +102,15 @@ public class DefaultServerPipelineFactory extends ServerPipelineFactory {
}
if (consumer.getConfiguration().isOrderedThreadPoolExecutor()) {
- // this must be added just before the ServerChannelHandler
- // use ordered thread pool, to ensure we process the events in order, and can send back
- // replies in the expected order. eg this is required by TCP.
- // and use a Camel thread factory so we have consistent thread namings
- ExecutionHandler executionHandler = new ExecutionHandler(consumer.getEndpoint().getComponent().getExecutorService());
- addToPipeline("executionHandler", channelPipeline, executionHandler);
- LOG.debug("Using OrderedMemoryAwareThreadPoolExecutor with core pool size: {}", consumer.getConfiguration().getMaximumPoolSize());
- }
-
- // our handler must be added last
- addToPipeline("handler", channelPipeline, new ServerChannelHandler(consumer));
+ // Just use EventExecutorGroup from the Netty Component
+ EventExecutorGroup applicationExecutor = consumer.getEndpoint().getComponent().getExecutorService();
+ addToPipeline("handler", channelPipeline, applicationExecutor, new ServerChannelHandler(consumer));
+
+ } else {
+ // still use the worker event loop group here
+ addToPipeline("handler", channelPipeline, new ServerChannelHandler(consumer));
+ }
LOG.trace("Created ChannelPipeline: {}", channelPipeline);
}
@@ -120,6 +118,10 @@ public class DefaultServerPipelineFactory extends ServerPipelineFactory {
private void addToPipeline(String name, ChannelPipeline pipeline, ChannelHandler handler) {
pipeline.addLast(name, handler);
}
+
+ private void addToPipeline(String name, ChannelPipeline pipeline, EventExecutorGroup executor, ChannelHandler handler) {
+ pipeline.addLast(executor, name, handler);
+ }
private SSLContext createSSLContext(CamelContext camelContext, NettyServerBootstrapConfiguration configuration) throws Exception {
if (!configuration.isSsl()) {
http://git-wip-us.apache.org/repos/asf/camel/blob/7c89323a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyComponent.java
----------------------------------------------------------------------
diff --git a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyComponent.java b/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyComponent.java
index 72dfd7e..f23431a 100644
--- a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyComponent.java
+++ b/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyComponent.java
@@ -20,22 +20,23 @@ import java.net.URI;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ThreadFactory;
-import java.util.concurrent.TimeUnit;
+
+import io.netty.util.HashedWheelTimer;
+import io.netty.util.Timer;
+import io.netty.util.concurrent.DefaultEventExecutorGroup;
+import io.netty.util.concurrent.EventExecutorGroup;
import org.apache.camel.CamelContext;
import org.apache.camel.Endpoint;
import org.apache.camel.impl.UriEndpointComponent;
import org.apache.camel.util.IntrospectionSupport;
import org.apache.camel.util.concurrent.CamelThreadFactory;
-import io.netty.handler.execution.OrderedMemoryAwareThreadPoolExecutor;
-import io.netty.util.HashedWheelTimer;
-import io.netty.util.Timer;
public class NettyComponent extends UriEndpointComponent {
// use a shared timer for Netty (see javadoc for HashedWheelTimer)
private static volatile Timer timer;
private NettyConfiguration configuration;
- private OrderedMemoryAwareThreadPoolExecutor executorService;
+ private volatile EventExecutorGroup executorService;
public NettyComponent() {
super(NettyEndpoint.class);
@@ -99,7 +100,7 @@ public class NettyComponent extends UriEndpointComponent {
return timer;
}
- public synchronized OrderedMemoryAwareThreadPoolExecutor getExecutorService() {
+ public synchronized EventExecutorGroup getExecutorService() {
if (executorService == null) {
executorService = createExecutorService();
}
@@ -122,15 +123,13 @@ public class NettyComponent extends UriEndpointComponent {
super.doStart();
}
- protected OrderedMemoryAwareThreadPoolExecutor createExecutorService() {
- // use ordered thread pool, to ensure we process the events in order, and can send back
- // replies in the expected order. eg this is required by TCP.
+ protected EventExecutorGroup createExecutorService() {
+ // Provide the executor service for the application
// and use a Camel thread factory so we have consistent thread namings
// we should use a shared thread pool as recommended by Netty
String pattern = getCamelContext().getExecutorServiceManager().getThreadNamePattern();
- ThreadFactory factory = new CamelThreadFactory(pattern, "NettyOrderedWorker", true);
- return new OrderedMemoryAwareThreadPoolExecutor(configuration.getMaximumPoolSize(),
- 0L, 0L, 30, TimeUnit.SECONDS, factory);
+ ThreadFactory factory = new CamelThreadFactory(pattern, "NettyEventExecutorGroup", true);
+ return new DefaultEventExecutorGroup(configuration.getMaximumPoolSize(), factory);
}
@Override
http://git-wip-us.apache.org/repos/asf/camel/blob/7c89323a/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 639ec88..47b8af5 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
@@ -79,6 +79,7 @@ public class NettyConfiguration extends NettyServerBootstrapConfiguration implem
@UriParam
private int maximumPoolSize = 16;
@UriParam
+ // TODO we need to rename this property
private boolean orderedThreadPoolExecutor = true;
@UriParam
private int producerPoolMaxActive = -1;
@@ -197,7 +198,7 @@ public class NettyConfiguration extends NettyServerBootstrapConfiguration implem
decoders.add(ChannelHandlerFactories.newStringDecoder(charset));
if (LOG.isDebugEnabled()) {
- LOG.debug("Using textline encoders and decoders with charset: {}, delimiter: {} and decoderMaxLineLength: {}",
+ LOG.debug("Using textline encoders and decoders with charset: {}, delimiter: {} and decoderMaxLineLength: {}",
new Object[]{charset, delimiter, decoderMaxLineLength});
}
} else {
http://git-wip-us.apache.org/repos/asf/camel/blob/7c89323a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyEndpoint.java b/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyEndpoint.java
index 3487ef7..79d30e4 100644
--- a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyEndpoint.java
+++ b/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyEndpoint.java
@@ -109,10 +109,10 @@ public class NettyEndpoint extends DefaultEndpoint {
}
protected SSLSession getSSLSession(ChannelHandlerContext ctx) {
- final SslHandler sslHandler = ctx.getPipeline().get(SslHandler.class);
+ final SslHandler sslHandler = ctx.pipeline().get(SslHandler.class);
SSLSession sslSession = null;
if (sslHandler != null) {
- sslSession = sslHandler.getEngine().getSession();
+ sslSession = sslHandler.engine().getSession();
}
return sslSession;
}
http://git-wip-us.apache.org/repos/asf/camel/blob/7c89323a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyProducer.java b/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyProducer.java
index 2a02031..c94a4dc 100644
--- a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyProducer.java
+++ b/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyProducer.java
@@ -19,6 +19,7 @@ package org.apache.camel.component.netty4;
import java.net.ConnectException;
import java.net.InetSocketAddress;
import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.RejectedExecutionException;
@@ -42,6 +43,7 @@ import io.netty.channel.Channel;
import io.netty.channel.ChannelFactory;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.EventLoopGroup;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.ChannelGroupFuture;
import io.netty.channel.group.DefaultChannelGroup;
@@ -53,22 +55,24 @@ import io.netty.channel.socket.nio.NioDatagramWorkerPool;
import io.netty.channel.socket.nio.WorkerPool;
import io.netty.util.HashedWheelTimer;
import io.netty.util.Timer;
+import io.netty.util.concurrent.ImmediateEventExecutor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class NettyProducer extends DefaultAsyncProducer {
private static final Logger LOG = LoggerFactory.getLogger(NettyProducer.class);
- private final ChannelGroup allChannels = new DefaultChannelGroup("NettyProducer");
+ private final ChannelGroup allChannels = new DefaultChannelGroup("NettyProducer", ImmediateEventExecutor.INSTANCE);
private CamelContext context;
private NettyConfiguration configuration;
private ChannelFactory channelFactory;
private DatagramChannelFactory datagramChannelFactory;
private ClientPipelineFactory pipelineFactory;
private CamelLogger noReplyLogger;
- private BossPool bossPool;
- private WorkerPool workerPool;
+ private EventLoopGroup bossGroup;
+ private EventLoopGroup workerGroup;
private ObjectPool<Channel> pool;
private Timer timer;
+ private Map<Channel, NettyCamelState> nettyCamelStatesMap = new ConcurrentHashMap<Channel, NettyCamelState>();
public NettyProducer(NettyEndpoint nettyEndpoint, NettyConfiguration configuration) {
super(nettyEndpoint);
@@ -157,19 +161,14 @@ public class NettyProducer extends DefaultAsyncProducer {
ChannelGroupFuture future = allChannels.close();
future.awaitUninterruptibly();
- // and then release other resources
- if (channelFactory != null) {
- channelFactory.releaseExternalResources();
- }
-
// and then shutdown the thread pools
- if (bossPool != null) {
- bossPool.shutdown();
- bossPool = null;
+ if (bossGroup != null) {
+ bossGroup.shutdownGracefully();
+ bossGroup = null;
}
- if (workerPool != null) {
- workerPool.shutdown();
- workerPool = null;
+ if (workerGroup != null) {
+ workerGroup.shutdownGracefully();
+ workerGroup = null;
}
if (pool != null) {
@@ -245,7 +244,7 @@ public class NettyProducer extends DefaultAsyncProducer {
final AsyncCallback producerCallback = new NettyProducerCallback(channel, callback);
// setup state as attachment on the channel, so we can access the state later when needed
- channel.setAttachment(new NettyCamelState(producerCallback, exchange));
+ putState(channel, new NettyCamelState(producerCallback, exchange));
// write body
NettyHelper.writeBodyAsync(LOG, channel, null, body, exchange, new ChannelFutureListener() {
@@ -317,7 +316,7 @@ public class NettyProducer extends DefaultAsyncProducer {
* To get the {@link NettyCamelState} from the given channel.
*/
public NettyCamelState getState(Channel channel) {
- return (NettyCamelState) channel.getAttachment();
+ return nettyCamelStatesMap.get(channel);
}
/**
@@ -325,7 +324,14 @@ public class NettyProducer extends DefaultAsyncProducer {
* when no longer needed
*/
public void removeState(Channel channel) {
- channel.setAttachment(null);
+ nettyCamelStatesMap.remove(channel);
+ }
+
+ /**
+ * Put the {@link NettyCamelState} into the map use the given channel as the key
+ */
+ public void putState(Channel channel, NettyCamelState state) {
+ nettyCamelStatesMap.put(channel, state);
}
protected void setupTCPCommunication() throws Exception {
@@ -451,7 +457,7 @@ public class NettyProducer extends DefaultAsyncProducer {
}
throw cause;
}
- Channel answer = channelFuture.getChannel();
+ Channel answer = channelFuture.channel();
// to keep track of all channels in use
allChannels.add(answer);
@@ -532,7 +538,7 @@ public class NettyProducer extends DefaultAsyncProducer {
@Override
public boolean validateObject(Channel channel) {
// we need a connected channel to be valid
- boolean answer = channel.isConnected();
+ boolean answer = channel.isActive();
LOG.trace("Validating channel: {} -> {}", channel, answer);
return answer;
}
http://git-wip-us.apache.org/repos/asf/camel/blob/7c89323a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyServerBootstrapConfiguration.java
----------------------------------------------------------------------
diff --git a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyServerBootstrapConfiguration.java b/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyServerBootstrapConfiguration.java
index eccae1f..d862b59 100644
--- a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyServerBootstrapConfiguration.java
+++ b/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyServerBootstrapConfiguration.java
@@ -19,11 +19,11 @@ package org.apache.camel.component.netty4;
import java.io.File;
import java.util.Map;
-import org.apache.camel.util.jsse.SSLContextParameters;
-import io.netty.channel.socket.nio.BossPool;
-import io.netty.channel.socket.nio.WorkerPool;
+import io.netty.channel.EventLoopGroup;
import io.netty.handler.ssl.SslHandler;
+import org.apache.camel.util.jsse.SSLContextParameters;
+
public class NettyServerBootstrapConfiguration implements Cloneable {
protected String protocol;
@@ -38,7 +38,7 @@ public class NettyServerBootstrapConfiguration implements Cloneable {
protected boolean keepAlive = true;
protected boolean tcpNoDelay = true;
protected boolean reuseAddress = true;
- protected long connectTimeout = 10000;
+ protected int connectTimeout = 10000;
protected int backlog;
protected ServerPipelineFactory serverPipelineFactory;
protected NettyServerBootstrapFactory nettyServerBootstrapFactory;
@@ -56,8 +56,8 @@ public class NettyServerBootstrapConfiguration implements Cloneable {
protected String keyStoreFormat;
protected String securityProvider;
protected String passphrase;
- protected BossPool bossPool;
- protected WorkerPool workerPool;
+ protected EventLoopGroup bossGroup;
+ protected EventLoopGroup workerGroup;
protected String networkInterface;
public String getAddress() {
@@ -164,11 +164,11 @@ public class NettyServerBootstrapConfiguration implements Cloneable {
this.reuseAddress = reuseAddress;
}
- public long getConnectTimeout() {
+ public int getConnectTimeout() {
return connectTimeout;
}
- public void setConnectTimeout(long connectTimeout) {
+ public void setConnectTimeout(int connectTimeout) {
this.connectTimeout = connectTimeout;
}
@@ -304,20 +304,20 @@ public class NettyServerBootstrapConfiguration implements Cloneable {
this.options = options;
}
- public BossPool getBossPool() {
- return bossPool;
+ public EventLoopGroup getBossGroup() {
+ return bossGroup;
}
- public void setBossPool(BossPool bossPool) {
- this.bossPool = bossPool;
+ public void setBossPool(EventLoopGroup bossGroup) {
+ this.bossGroup = bossGroup;
}
- public WorkerPool getWorkerPool() {
- return workerPool;
+ public EventLoopGroup getWorkerGroup() {
+ return workerGroup;
}
- public void setWorkerPool(WorkerPool workerPool) {
- this.workerPool = workerPool;
+ public void setWorkerPool(EventLoopGroup workerGroup) {
+ this.workerGroup = workerGroup;
}
public String getNetworkInterface() {
@@ -401,9 +401,9 @@ public class NettyServerBootstrapConfiguration implements Cloneable {
isCompatible = false;
} else if (passphrase != null && !passphrase.equals(other.passphrase)) {
isCompatible = false;
- } else if (bossPool != other.bossPool) {
+ } else if (bossGroup != other.bossGroup) {
isCompatible = false;
- } else if (workerPool != other.workerPool) {
+ } else if (workerGroup != other.workerGroup) {
isCompatible = false;
} else if (networkInterface != null && !networkInterface.equals(other.networkInterface)) {
isCompatible = false;
@@ -442,8 +442,8 @@ public class NettyServerBootstrapConfiguration implements Cloneable {
+ ", keyStoreFormat='" + keyStoreFormat + '\''
+ ", securityProvider='" + securityProvider + '\''
+ ", passphrase='" + passphrase + '\''
- + ", bossPool=" + bossPool
- + ", workerPool=" + workerPool
+ + ", bossGroup=" + bossGroup
+ + ", workerGroup=" + workerGroup
+ ", networkInterface='" + networkInterface + '\''
+ '}';
}
http://git-wip-us.apache.org/repos/asf/camel/blob/7c89323a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyServerBootstrapFactory.java
----------------------------------------------------------------------
diff --git a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyServerBootstrapFactory.java b/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyServerBootstrapFactory.java
index ea461ab..e2c6865 100644
--- a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyServerBootstrapFactory.java
+++ b/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyServerBootstrapFactory.java
@@ -15,13 +15,13 @@
* limitations under the License.
*/
package org.apache.camel.component.netty4;
-
import java.util.concurrent.ThreadFactory;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelInitializer;
+
import org.apache.camel.CamelContext;
import org.apache.camel.SuspendableService;
-import io.netty.channel.Channel;
-import io.netty.channel.ChannelPipelineFactory;
/**
* Factory for setting up Netty {@link io.netty.bootstrap.ServerBootstrap} and all
@@ -37,18 +37,18 @@ public interface NettyServerBootstrapFactory extends SuspendableService {
*
* @param camelContext the {@link CamelContext} for non-shared bootstrap factory
* @param configuration the bootstrap configuration
- * @param pipelineFactory the pipeline factory
+ * @param pipelineFactory the channel initializer which set up the channel handler pipeline
*/
- void init(CamelContext camelContext, NettyServerBootstrapConfiguration configuration, ChannelPipelineFactory pipelineFactory);
+ void init(CamelContext camelContext, NettyServerBootstrapConfiguration configuration, ChannelInitializer<Channel> pipelineFactory);
/**
* Initializes this <b>shared</b> {@link NettyServerBootstrapFactory}.
*
* @param threadFactory the thread factory to use for shared bootstrap factory
* @param configuration the bootstrap configuration
- * @param pipelineFactory the pipeline factory
+ * @param pipelineFactory the channel initializer which set up the channel handler pipeline
*/
- void init(ThreadFactory threadFactory, NettyServerBootstrapConfiguration configuration, ChannelPipelineFactory pipelineFactory);
+ void init(ThreadFactory threadFactory, NettyServerBootstrapConfiguration configuration, ChannelInitializer<Channel> pipelineFactory);
/**
* When a new {@link Channel} is opened.
http://git-wip-us.apache.org/repos/asf/camel/blob/7c89323a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyServerBossPoolBuilder.java
----------------------------------------------------------------------
diff --git a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyServerBossPoolBuilder.java b/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyServerBossPoolBuilder.java
index 469105c..395d8f4 100644
--- a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyServerBossPoolBuilder.java
+++ b/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyServerBossPoolBuilder.java
@@ -16,13 +16,13 @@
*/
package org.apache.camel.component.netty4;
-import java.util.concurrent.Executors;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import org.apache.camel.util.concurrent.CamelThreadFactory;
-import io.netty.channel.socket.nio.BossPool;
-import io.netty.channel.socket.nio.NioServerBossPool;
/**
- * A builder to create Netty {@link EventLoopGroup } which can be used for sharing boss pools
+ * A builder to create Netty {@link io.netty.channel.EventLoopGroup} which can be used for executor boss events
* with multiple Netty {@link org.apache.camel.component.netty4.NettyServerBootstrapFactory} server bootstrap configurations.
*/
public final class NettyServerBossPoolBuilder {
@@ -61,7 +61,7 @@ public final class NettyServerBossPoolBuilder {
/**
* Creates a new boss pool.
*/
- BossPool build() {
- return new NioServerBossPool(Executors.newCachedThreadPool(), bossCount, new CamelNettyThreadNameDeterminer(pattern, name));
+ EventLoopGroup build() {
+ return new NioEventLoopGroup(bossCount, new CamelThreadFactory(pattern, name, false));
}
}
http://git-wip-us.apache.org/repos/asf/camel/blob/7c89323a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyWorkerPoolBuilder.java
----------------------------------------------------------------------
diff --git a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyWorkerPoolBuilder.java b/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyWorkerPoolBuilder.java
index 409d1eb..125bbce 100644
--- a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyWorkerPoolBuilder.java
+++ b/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/NettyWorkerPoolBuilder.java
@@ -16,13 +16,12 @@
*/
package org.apache.camel.component.netty4;
-import java.util.concurrent.Executors;
-
-import io.netty.channel.socket.nio.NioWorkerPool;
-import io.netty.channel.socket.nio.WorkerPool;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import org.apache.camel.util.concurrent.CamelThreadFactory;
/**
- * A builder to create Netty {@link WorkerPool} which can be used for sharing worker pools
+ * A builder to create Netty {@link io.netty.channel.EventLoopGroup} which can be used for sharing worker pools
* with multiple Netty {@link NettyServerBootstrapFactory} server bootstrap configurations.
*/
public final class NettyWorkerPoolBuilder {
@@ -30,7 +29,7 @@ public final class NettyWorkerPoolBuilder {
private String name = "NettyWorker";
private String pattern;
private int workerCount;
- private volatile WorkerPool workerPool;
+ private volatile EventLoopGroup workerPool;
public void setName(String name) {
this.name = name;
@@ -62,9 +61,9 @@ public final class NettyWorkerPoolBuilder {
/**
* Creates a new worker pool.
*/
- public WorkerPool build() {
+ public EventLoopGroup build() {
int count = workerCount > 0 ? workerCount : NettyHelper.DEFAULT_IO_THREADS;
- workerPool = new NioWorkerPool(Executors.newCachedThreadPool(), count, new CamelNettyThreadNameDeterminer(pattern, name));
+ workerPool = new NioEventLoopGroup(count, new CamelThreadFactory(pattern, name, false));
return workerPool;
}
@@ -73,7 +72,7 @@ public final class NettyWorkerPoolBuilder {
*/
public void destroy() {
if (workerPool != null) {
- workerPool.shutdown();
+ workerPool.shutdownGracefully();
workerPool = null;
}
}
http://git-wip-us.apache.org/repos/asf/camel/blob/7c89323a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/SingleTCPNettyServerBootstrapFactory.java
----------------------------------------------------------------------
diff --git a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/SingleTCPNettyServerBootstrapFactory.java b/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/SingleTCPNettyServerBootstrapFactory.java
index 468bc32..6df8b06 100644
--- a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/SingleTCPNettyServerBootstrapFactory.java
+++ b/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/SingleTCPNettyServerBootstrapFactory.java
@@ -20,19 +20,20 @@ import java.net.InetSocketAddress;
import java.util.Map;
import java.util.concurrent.ThreadFactory;
-import org.apache.camel.CamelContext;
-import org.apache.camel.support.ServiceSupport;
+import io.netty.bootstrap.ChannelFactory;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
-import io.netty.channel.ChannelFactory;
import io.netty.channel.ChannelFuture;
-import io.netty.channel.ChannelPipelineFactory;
+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.ChannelGroupFuture;
import io.netty.channel.group.DefaultChannelGroup;
-import io.netty.channel.socket.nio.BossPool;
-import io.netty.channel.socket.nio.NioServerSocketChannelFactory;
-import io.netty.channel.socket.nio.WorkerPool;
+import io.netty.channel.socket.nio.NioServerSocketChannel;
+import io.netty.util.concurrent.ImmediateEventExecutor;
+import org.apache.camel.CamelContext;
+import org.apache.camel.support.ServiceSupport;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -46,24 +47,24 @@ public class SingleTCPNettyServerBootstrapFactory extends ServiceSupport impleme
private CamelContext camelContext;
private ThreadFactory threadFactory;
private NettyServerBootstrapConfiguration configuration;
- private ChannelPipelineFactory pipelineFactory;
- private ChannelFactory channelFactory;
+ private ChannelInitializer<Channel> pipelineFactory;
private ServerBootstrap serverBootstrap;
private Channel channel;
- private BossPool bossPool;
- private WorkerPool workerPool;
+ private EventLoopGroup bossGroup;
+ private EventLoopGroup workerGroup;
public SingleTCPNettyServerBootstrapFactory() {
- this.allChannels = new DefaultChannelGroup(SingleTCPNettyServerBootstrapFactory.class.getName());
+ // The executor just execute tasks in the callers thread
+ this.allChannels = new DefaultChannelGroup(SingleTCPNettyServerBootstrapFactory.class.getName(), ImmediateEventExecutor.INSTANCE);
}
- public void init(CamelContext camelContext, NettyServerBootstrapConfiguration configuration, ChannelPipelineFactory pipelineFactory) {
+ 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, ChannelPipelineFactory pipelineFactory) {
+ public void init(ThreadFactory threadFactory, NettyServerBootstrapConfiguration configuration, ChannelInitializer<Channel> pipelineFactory) {
this.threadFactory = threadFactory;
this.configuration = configuration;
this.pipelineFactory = pipelineFactory;
@@ -124,52 +125,55 @@ public class SingleTCPNettyServerBootstrapFactory extends ServiceSupport impleme
protected void startServerBootstrap() {
// prefer using explicit configured thread pools
- BossPool bp = configuration.getBossPool();
- WorkerPool wp = configuration.getWorkerPool();
+ EventLoopGroup bg = configuration.getBossGroup();
+ EventLoopGroup wg = configuration.getWorkerGroup();
- if (bp == null) {
+ if (bg == null) {
// create new pool which we should shutdown when stopping as its not shared
- bossPool = new NettyServerBossPoolBuilder()
+ bossGroup = new NettyServerBossPoolBuilder()
.withBossCount(configuration.getBossCount())
.withName("NettyServerTCPBoss")
.build();
- bp = bossPool;
+ bg = bossGroup;
}
- if (wp == null) {
+ if (wg == null) {
// create new pool which we should shutdown when stopping as its not shared
- workerPool = new NettyWorkerPoolBuilder()
+ workerGroup = new NettyWorkerPoolBuilder()
.withWorkerCount(configuration.getWorkerCount())
.withName("NettyServerTCPWorker")
.build();
- wp = workerPool;
+ wg = workerGroup;
}
- channelFactory = new NioServerSocketChannelFactory(bp, wp);
+ //channelFactory = new NioServerSocketChannelFactory(bg, wg);
- serverBootstrap = new ServerBootstrap(channelFactory);
- serverBootstrap.setOption("child.keepAlive", configuration.isKeepAlive());
- serverBootstrap.setOption("child.tcpNoDelay", configuration.isTcpNoDelay());
- serverBootstrap.setOption("reuseAddress", configuration.isReuseAddress());
- serverBootstrap.setOption("child.reuseAddress", configuration.isReuseAddress());
- serverBootstrap.setOption("child.connectTimeoutMillis", configuration.getConnectTimeout());
+ serverBootstrap = new ServerBootstrap();
+ serverBootstrap.group(bg, wg).channel(NioServerSocketChannel.class);
+ serverBootstrap.childOption(ChannelOption.SO_KEEPALIVE, configuration.isKeepAlive());
+ serverBootstrap.childOption(ChannelOption.TCP_NODELAY, configuration.isTcpNoDelay());
+ serverBootstrap.option(ChannelOption.SO_REUSEADDR, configuration.isReuseAddress());
+ serverBootstrap.childOption(ChannelOption.SO_REUSEADDR, configuration.isReuseAddress());
+ serverBootstrap.childOption(ChannelOption.CONNECT_TIMEOUT_MILLIS, configuration.getConnectTimeout());
if (configuration.getBacklog() > 0) {
- serverBootstrap.setOption("backlog", configuration.getBacklog());
+ serverBootstrap.option(ChannelOption.SO_BACKLOG, configuration.getBacklog());
}
- // set any additional netty options
- if (configuration.getOptions() != null) {
+ // TODO set any additional netty options and child options
+ /*if (configuration.getOptions() != null) {
for (Map.Entry<String, Object> entry : configuration.getOptions().entrySet()) {
serverBootstrap.setOption(entry.getKey(), entry.getValue());
}
- }
-
- LOG.debug("Created ServerBootstrap {} with options: {}", serverBootstrap, serverBootstrap.getOptions());
+ }*/
// set the pipeline factory, which creates the pipeline for each newly created channels
- serverBootstrap.setPipelineFactory(pipelineFactory);
+ serverBootstrap.handler(pipelineFactory);
+
+ LOG.debug("Created ServerBootstrap {}", serverBootstrap);
LOG.info("ServerBootstrap binding to {}:{}", configuration.getHost(), configuration.getPort());
- channel = serverBootstrap.bind(new InetSocketAddress(configuration.getHost(), configuration.getPort()));
+ ChannelFuture channelFutrue = serverBootstrap.bind(new InetSocketAddress(configuration.getHost(), configuration.getPort()));
+ // TODO how can we make sure the server is bind rightly
+ channel = channelFutrue.channel();
// to keep track of all channels in use
allChannels.add(channel);
}
@@ -182,20 +186,18 @@ public class SingleTCPNettyServerBootstrapFactory extends ServiceSupport impleme
ChannelGroupFuture future = allChannels.close();
future.awaitUninterruptibly();
- // close server external resources
- if (channelFactory != null) {
- channelFactory.releaseExternalResources();
- channelFactory = null;
+ if (allChannels != null) {
+ allChannels.close();
}
// and then shutdown the thread pools
- if (bossPool != null) {
- bossPool.shutdown();
- bossPool = null;
+ if (bossGroup != null) {
+ bossGroup.shutdownGracefully();
+ bossGroup = null;
}
- if (workerPool != null) {
- workerPool.shutdown();
- workerPool = null;
+ if (workerGroup != null) {
+ workerGroup.shutdownGracefully();
+ workerGroup = null;
}
}
http://git-wip-us.apache.org/repos/asf/camel/blob/7c89323a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/SingleUDPNettyServerBootstrapFactory.java
----------------------------------------------------------------------
diff --git a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/SingleUDPNettyServerBootstrapFactory.java b/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/SingleUDPNettyServerBootstrapFactory.java
index 04d2033..4f5bc8c 100644
--- a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/SingleUDPNettyServerBootstrapFactory.java
+++ b/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/SingleUDPNettyServerBootstrapFactory.java
@@ -16,6 +16,15 @@
*/
package org.apache.camel.component.netty4;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.group.ChannelGroup;
+import io.netty.channel.group.ChannelGroupFuture;
+import io.netty.channel.group.DefaultChannelGroup;
+import io.netty.channel.socket.DatagramChannel;
+import io.netty.util.concurrent.ImmediateEventExecutor;
+
import java.net.InetSocketAddress;
import java.net.NetworkInterface;
import java.util.Map;
@@ -53,25 +62,25 @@ public class SingleUDPNettyServerBootstrapFactory extends ServiceSupport impleme
private CamelContext camelContext;
private ThreadFactory threadFactory;
private NettyServerBootstrapConfiguration configuration;
- private ChannelPipelineFactory pipelineFactory;
+ private ChannelInitializer<Channel> pipelineFactory;
private DatagramChannelFactory datagramChannelFactory;
private ConnectionlessBootstrap connectionlessBootstrap;
private NetworkInterface multicastNetworkInterface;
private DatagramChannel datagramChannel;
private Channel channel;
- private WorkerPool workerPool;
+ private EventLoopGroup workerGroup;
public SingleUDPNettyServerBootstrapFactory() {
- this.allChannels = new DefaultChannelGroup(SingleUDPNettyServerBootstrapFactory.class.getName());
+ this.allChannels = new DefaultChannelGroup(SingleUDPNettyServerBootstrapFactory.class.getName(), ImmediateEventExecutor.INSTANCE);
}
- public void init(CamelContext camelContext, NettyServerBootstrapConfiguration configuration, ChannelPipelineFactory pipelineFactory) {
+ 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, ChannelPipelineFactory pipelineFactory) {
+ public void init(ThreadFactory threadFactory, NettyServerBootstrapConfiguration configuration, ChannelInitializer<Channel> pipelineFactory) {
this.threadFactory = threadFactory;
this.configuration = configuration;
this.pipelineFactory = pipelineFactory;
@@ -186,9 +195,9 @@ public class SingleUDPNettyServerBootstrapFactory extends ServiceSupport impleme
}
// and then shutdown the thread pools
- if (workerPool != null) {
- workerPool.shutdown();
- workerPool = null;
+ if (workerGroup != null) {
+ workerGroup.
+ workerGroup = null;
}
}