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/22 15:52:59 UTC
[11/26] git commit: CAMEL-6555 Fixed the compile errors in
camel-netty4
CAMEL-6555 Fixed the compile errors in camel-netty4
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/fa1cdc43
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/fa1cdc43
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/fa1cdc43
Branch: refs/heads/master
Commit: fa1cdc43f742e7f1f8ed442907dbff2cd9d2d264
Parents: 66ccde6
Author: Willem Jiang <wi...@gmail.com>
Authored: Fri Jul 18 23:14:04 2014 +0800
Committer: Willem Jiang <wi...@gmail.com>
Committed: Tue Jul 22 21:25:18 2014 +0800
----------------------------------------------------------------------
.../netty4/DefaultServerPipelineFactory.java | 5 +-
.../SingleTCPNettyServerBootstrapFactory.java | 7 ++-
.../netty4/handlers/ClientChannelHandler.java | 7 +--
.../netty4/handlers/ServerChannelHandler.java | 2 +-
.../camel/component/netty4/MyCustomCodec.java | 49 +++++++++++---------
5 files changed, 39 insertions(+), 31 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/fa1cdc43/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 cf9e57e..92c1b21 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
@@ -16,19 +16,18 @@
*/
package org.apache.camel.component.netty4;
-import java.nio.channels.Channels;
import java.util.List;
import javax.net.ssl.SSLContext;
import javax.net.ssl.SSLEngine;
-import org.apache.camel.CamelContext;
-import org.apache.camel.util.ObjectHelper;
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.CamelContext;
import org.apache.camel.component.netty4.handlers.ServerChannelHandler;
import org.apache.camel.component.netty4.ssl.SSLEngineFactory;
import org.apache.camel.util.ObjectHelper;
http://git-wip-us.apache.org/repos/asf/camel/blob/fa1cdc43/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 6df8b06..dbc668c 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
@@ -108,7 +108,9 @@ public class SingleTCPNettyServerBootstrapFactory extends ServiceSupport impleme
if (!future.isSuccess()) {
// if we cannot bind, the re-create channel
allChannels.remove(channel);
- channel = serverBootstrap.bind(new InetSocketAddress(configuration.getHost(), configuration.getPort()));
+ future = serverBootstrap.bind(new InetSocketAddress(configuration.getHost(), configuration.getPort()));
+ future.awaitUninterruptibly();
+ channel = future.channel();
allChannels.add(channel);
}
}
@@ -118,7 +120,8 @@ public class SingleTCPNettyServerBootstrapFactory extends ServiceSupport impleme
protected void doSuspend() throws Exception {
if (channel != null) {
LOG.debug("ServerBootstrap unbinding from {}:{}", configuration.getHost(), configuration.getPort());
- ChannelFuture future = channel.unbind();
+ //TODO need to check if it's good way to unbinding the channel
+ ChannelFuture future = channel.close();
future.awaitUninterruptibly();
}
}
http://git-wip-us.apache.org/repos/asf/camel/blob/fa1cdc43/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/handlers/ClientChannelHandler.java
----------------------------------------------------------------------
diff --git a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/handlers/ClientChannelHandler.java b/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/handlers/ClientChannelHandler.java
index c975afd..d4651a9 100644
--- a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/handlers/ClientChannelHandler.java
+++ b/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/handlers/ClientChannelHandler.java
@@ -16,6 +16,7 @@
*/
package org.apache.camel.component.netty4.handlers;
+import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import org.apache.camel.AsyncCallback;
@@ -122,14 +123,14 @@ public class ClientChannelHandler extends SimpleChannelInboundHandler<Object> {
messageReceived = true;
if (LOG.isTraceEnabled()) {
- LOG.trace("Message received: {}", messageEvent);
+ LOG.trace("Message received: {}", msg);
}
if (producer.getConfiguration().getRequestTimeout() > 0) {
- ChannelHandler handler = ctx.getPipeline().get("timeout");
+ ChannelHandler handler = ctx.pipeline().get("timeout");
if (handler != null) {
LOG.trace("Removing timeout channel as we received message");
- ctx.getPipeline().remove(handler);
+ ctx.pipeline().remove(handler);
}
}
http://git-wip-us.apache.org/repos/asf/camel/blob/fa1cdc43/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/handlers/ServerChannelHandler.java
----------------------------------------------------------------------
diff --git a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/handlers/ServerChannelHandler.java b/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/handlers/ServerChannelHandler.java
index d800b5d..47dce7b 100644
--- a/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/handlers/ServerChannelHandler.java
+++ b/components/camel-netty4/src/main/java/org/apache/camel/component/netty4/handlers/ServerChannelHandler.java
@@ -96,7 +96,7 @@ public class ServerChannelHandler extends SimpleChannelInboundHandler<Object> {
// we want to handle the UoW
consumer.createUoW(exchange);
- beforeProcess(exchange, messageEvent);
+ beforeProcess(exchange, ctx, msg);
// process accordingly to endpoint configuration
if (consumer.getEndpoint().isSynchronous()) {
http://git-wip-us.apache.org/repos/asf/camel/blob/fa1cdc43/components/camel-netty4/src/test/java/org/apache/camel/component/netty4/MyCustomCodec.java
----------------------------------------------------------------------
diff --git a/components/camel-netty4/src/test/java/org/apache/camel/component/netty4/MyCustomCodec.java b/components/camel-netty4/src/test/java/org/apache/camel/component/netty4/MyCustomCodec.java
index ee8ba36..08a9ebc 100644
--- a/components/camel-netty4/src/test/java/org/apache/camel/component/netty4/MyCustomCodec.java
+++ b/components/camel-netty4/src/test/java/org/apache/camel/component/netty4/MyCustomCodec.java
@@ -16,26 +16,26 @@
*/
package org.apache.camel.component.netty4;
-import org.jboss.netty.buffer.ChannelBuffer;
-import org.jboss.netty.buffer.ChannelBuffers;
-import org.jboss.netty.channel.Channel;
-import org.jboss.netty.channel.ChannelHandler;
-import org.jboss.netty.channel.ChannelHandlerContext;
-import org.jboss.netty.handler.codec.oneone.OneToOneDecoder;
-import org.jboss.netty.handler.codec.oneone.OneToOneEncoder;
+import java.util.List;
-import static org.jboss.netty.buffer.ChannelBuffers.copiedBuffer;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufAllocator;
+import io.netty.buffer.EmptyByteBuf;
+import io.netty.channel.ChannelHandler;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.MessageToMessageDecoder;
+import io.netty.handler.codec.MessageToMessageEncoder;
public final class MyCustomCodec {
- private static ChannelBuffer nullDelimiter = ChannelBuffers.wrappedBuffer(new byte[]{0});
+ private static ByteBuf nullDelimiter = new EmptyByteBuf(ByteBufAllocator.DEFAULT);
private MyCustomCodec() {
// Helper class
}
public static ChannelHandlerFactory createMyCustomDecoder() {
- ChannelBuffer[] delimiters = new ChannelBuffer[]{nullDelimiter, nullDelimiter};
+ ByteBuf[] delimiters = new ByteBuf[]{nullDelimiter, nullDelimiter};
return ChannelHandlerFactories.newDelimiterBasedFrameDecoder(4096, delimiters);
}
@@ -48,34 +48,39 @@ public final class MyCustomCodec {
}
@ChannelHandler.Sharable
- public static class BytesDecoder extends OneToOneDecoder {
+ public static class BytesDecoder extends MessageToMessageDecoder<Object> {
@Override
- protected Object decode(ChannelHandlerContext ctx, Channel channel, Object msg) throws Exception {
- if (!(msg instanceof ChannelBuffer)) {
- return msg;
+ protected void decode(ChannelHandlerContext ctx, Object msg, List<Object> out) throws Exception {
+ if (!(msg instanceof ByteBuf)) {
+ out.add(msg);
} else {
// it may be empty, then return null
- ChannelBuffer cb = (ChannelBuffer) msg;
- if (cb.hasArray() && cb.readable()) {
- return cb.array();
+ ByteBuf cb = (ByteBuf) msg;
+ if (cb.hasArray() && cb.isReadable()) {
+ out.add(cb.array());
} else {
- return null;
+ out.add((Object)null);
}
}
+
}
}
@ChannelHandler.Sharable
- public static class BytesEncoder extends OneToOneEncoder {
+ public static class BytesEncoder extends MessageToMessageEncoder<Object> {
@Override
- protected Object encode(ChannelHandlerContext ctx, Channel channel, Object msg) throws Exception {
+ protected void encode(ChannelHandlerContext ctx, Object msg, List<Object> out) throws Exception {
if (msg instanceof byte[]) {
- return copiedBuffer((byte[]) msg);
+ byte[] bytes = (byte[])msg;
+ ByteBuf buf = ByteBufAllocator.DEFAULT.buffer(bytes.length);
+ buf.writeBytes(bytes);
+ out.add(buf);
+ } else {
+ out.add(msg);
}
- return msg;
}
}
}