From e846505ceb4bac5bc9472ed141ffe950418c29c2 Mon Sep 17 00:00:00 2001 From: Trustin Lee Date: Sun, 20 May 2012 12:53:22 +0900 Subject: [PATCH] Add combined codec classes and related utility method --- .../handler/codec/MessageToMessageCodec.java | 51 +++++++ .../handler/codec/StreamToMessageCodec.java | 61 ++++++++ .../handler/codec/StreamToStreamCodec.java | 60 ++++++++ .../netty/channel/ChannelHandlerAdapter.java | 140 ++++++++++++++++++ 4 files changed, 312 insertions(+) create mode 100644 codec/src/main/java/io/netty/handler/codec/MessageToMessageCodec.java create mode 100644 codec/src/main/java/io/netty/handler/codec/StreamToMessageCodec.java create mode 100644 codec/src/main/java/io/netty/handler/codec/StreamToStreamCodec.java diff --git a/codec/src/main/java/io/netty/handler/codec/MessageToMessageCodec.java b/codec/src/main/java/io/netty/handler/codec/MessageToMessageCodec.java new file mode 100644 index 0000000000..05745d563b --- /dev/null +++ b/codec/src/main/java/io/netty/handler/codec/MessageToMessageCodec.java @@ -0,0 +1,51 @@ +package io.netty.handler.codec; + +import io.netty.channel.ChannelBufferHolder; +import io.netty.channel.ChannelFuture; +import io.netty.channel.ChannelHandlerAdapter; +import io.netty.channel.ChannelInboundHandlerContext; +import io.netty.channel.ChannelOutboundHandlerContext; + +public abstract class MessageToMessageCodec + extends ChannelHandlerAdapter { + + private final MessageToMessageEncoder encoder = + new MessageToMessageEncoder() { + @Override + public OUTBOUND_OUT encode(ChannelOutboundHandlerContext ctx, OUTBOUND_IN msg) throws Exception { + return MessageToMessageCodec.this.encode(ctx, msg); + } + }; + + private final MessageToMessageDecoder decoder = + new MessageToMessageDecoder() { + @Override + public INBOUND_OUT decode(ChannelInboundHandlerContext ctx, INBOUND_IN msg) throws Exception { + return MessageToMessageCodec.this.decode(ctx, msg); + } + }; + + @Override + public ChannelBufferHolder newInboundBuffer(ChannelInboundHandlerContext ctx) throws Exception { + return decoder.newInboundBuffer(ctx); + } + + @Override + public void inboundBufferUpdated( + ChannelInboundHandlerContext ctx) throws Exception { + decoder.inboundBufferUpdated(ctx); + } + + @Override + public ChannelBufferHolder newOutboundBuffer(ChannelOutboundHandlerContext ctx) throws Exception { + return encoder.newOutboundBuffer(ctx); + } + + @Override + public void flush(ChannelOutboundHandlerContext ctx, ChannelFuture future) throws Exception { + encoder.flush(ctx, future); + } + + public abstract OUTBOUND_OUT encode(ChannelOutboundHandlerContext ctx, OUTBOUND_IN msg) throws Exception; + public abstract INBOUND_OUT decode(ChannelInboundHandlerContext ctx, INBOUND_IN msg) throws Exception; +} diff --git a/codec/src/main/java/io/netty/handler/codec/StreamToMessageCodec.java b/codec/src/main/java/io/netty/handler/codec/StreamToMessageCodec.java new file mode 100644 index 0000000000..8781f9f321 --- /dev/null +++ b/codec/src/main/java/io/netty/handler/codec/StreamToMessageCodec.java @@ -0,0 +1,61 @@ +package io.netty.handler.codec; + +import io.netty.buffer.ChannelBuffer; +import io.netty.channel.ChannelBufferHolder; +import io.netty.channel.ChannelFuture; +import io.netty.channel.ChannelHandlerAdapter; +import io.netty.channel.ChannelInboundHandlerContext; +import io.netty.channel.ChannelOutboundHandlerContext; + +public abstract class StreamToMessageCodec + extends ChannelHandlerAdapter { + + private final MessageToStreamEncoder encoder = + new MessageToStreamEncoder() { + @Override + public void encode( + ChannelOutboundHandlerContext ctx, + OUTBOUND_IN msg, ChannelBuffer out) throws Exception { + StreamToMessageCodec.this.encode(ctx, msg, out); + } + }; + + private final StreamToMessageDecoder decoder = + new StreamToMessageDecoder() { + @Override + public INBOUND_OUT decode( + ChannelInboundHandlerContext ctx, ChannelBuffer in) throws Exception { + return StreamToMessageCodec.this.decode(ctx, in); + } + }; + + @Override + public ChannelBufferHolder newInboundBuffer( + ChannelInboundHandlerContext ctx) throws Exception { + return decoder.newInboundBuffer(ctx); + } + + @Override + public void inboundBufferUpdated(ChannelInboundHandlerContext ctx) throws Exception { + decoder.inboundBufferUpdated(ctx); + } + + @Override + public ChannelBufferHolder newOutboundBuffer( + ChannelOutboundHandlerContext ctx) throws Exception { + return encoder.newOutboundBuffer(ctx); + } + + @Override + public void flush( + ChannelOutboundHandlerContext ctx, ChannelFuture future) throws Exception { + encoder.flush(ctx, future); + } + + public abstract void encode( + ChannelOutboundHandlerContext ctx, + OUTBOUND_IN msg, ChannelBuffer out) throws Exception; + + public abstract INBOUND_OUT decode( + ChannelInboundHandlerContext ctx, ChannelBuffer in) throws Exception; +} diff --git a/codec/src/main/java/io/netty/handler/codec/StreamToStreamCodec.java b/codec/src/main/java/io/netty/handler/codec/StreamToStreamCodec.java new file mode 100644 index 0000000000..d35f9504f0 --- /dev/null +++ b/codec/src/main/java/io/netty/handler/codec/StreamToStreamCodec.java @@ -0,0 +1,60 @@ +package io.netty.handler.codec; + +import io.netty.buffer.ChannelBuffer; +import io.netty.channel.ChannelBufferHolder; +import io.netty.channel.ChannelFuture; +import io.netty.channel.ChannelHandlerAdapter; +import io.netty.channel.ChannelInboundHandlerContext; +import io.netty.channel.ChannelOutboundHandlerContext; + +public abstract class StreamToStreamCodec extends ChannelHandlerAdapter { + + private final StreamToStreamEncoder encoder = new StreamToStreamEncoder() { + @Override + public void encode( + ChannelOutboundHandlerContext ctx, + ChannelBuffer in, ChannelBuffer out) throws Exception { + StreamToStreamCodec.this.encode(ctx, in, out); + } + }; + + private final StreamToStreamDecoder decoder = new StreamToStreamDecoder() { + @Override + public void decode( + ChannelInboundHandlerContext ctx, + ChannelBuffer in, ChannelBuffer out) throws Exception { + StreamToStreamCodec.this.decode(ctx, in, out); + } + }; + + @Override + public ChannelBufferHolder newInboundBuffer( + ChannelInboundHandlerContext ctx) throws Exception { + return decoder.newInboundBuffer(ctx); + } + + @Override + public void inboundBufferUpdated(ChannelInboundHandlerContext ctx) throws Exception { + decoder.inboundBufferUpdated(ctx); + } + + @Override + public ChannelBufferHolder newOutboundBuffer( + ChannelOutboundHandlerContext ctx) throws Exception { + return encoder.newOutboundBuffer(ctx); + } + + @Override + public void flush( + ChannelOutboundHandlerContext ctx, ChannelFuture future) throws Exception { + encoder.flush(ctx, future); + } + + public abstract void encode( + ChannelOutboundHandlerContext ctx, + ChannelBuffer in, ChannelBuffer out) throws Exception; + + public abstract void decode( + ChannelInboundHandlerContext ctx, + ChannelBuffer in, ChannelBuffer out) throws Exception; +} diff --git a/transport/src/main/java/io/netty/channel/ChannelHandlerAdapter.java b/transport/src/main/java/io/netty/channel/ChannelHandlerAdapter.java index cd5d01b260..a6e40406a4 100644 --- a/transport/src/main/java/io/netty/channel/ChannelHandlerAdapter.java +++ b/transport/src/main/java/io/netty/channel/ChannelHandlerAdapter.java @@ -3,6 +3,146 @@ package io.netty.channel; import java.net.SocketAddress; public abstract class ChannelHandlerAdapter implements ChannelInboundHandler, ChannelOutboundHandler { + + public static ChannelHandlerAdapter combine( + ChannelInboundHandler inboundHandler, ChannelOutboundHandler outboundHandler) { + if (inboundHandler == null) { + throw new NullPointerException("inboundHandler"); + } + if (outboundHandler == null) { + throw new NullPointerException("outboundHandler"); + } + + final ChannelInboundHandler in = inboundHandler; + final ChannelOutboundHandler out = outboundHandler; + return new ChannelHandlerAdapter() { + + @Override + public ChannelBufferHolder newInboundBuffer( + ChannelInboundHandlerContext ctx) throws Exception { + return in.newInboundBuffer(ctx); + } + + @Override + public ChannelBufferHolder newOutboundBuffer( + ChannelOutboundHandlerContext ctx) throws Exception { + return out.newOutboundBuffer(ctx); + } + + @Override + public void beforeAdd(ChannelHandlerContext ctx) throws Exception { + try { + in.beforeAdd(ctx); + } finally { + out.beforeAdd(ctx); + } + } + + @Override + public void afterAdd(ChannelHandlerContext ctx) throws Exception { + try { + in.afterAdd(ctx); + } finally { + out.afterAdd(ctx); + } + } + + @Override + public void beforeRemove(ChannelHandlerContext ctx) throws Exception { + try { + in.beforeRemove(ctx); + } finally { + out.beforeRemove(ctx); + } + } + + @Override + public void afterRemove(ChannelHandlerContext ctx) throws Exception { + try { + in.afterRemove(ctx); + } finally { + out.afterRemove(ctx); + } + } + + @Override + public void channelRegistered(ChannelInboundHandlerContext ctx) throws Exception { + in.channelRegistered(ctx); + } + + @Override + public void channelUnregistered(ChannelInboundHandlerContext ctx) throws Exception { + in.channelUnregistered(ctx); + } + + @Override + public void channelActive(ChannelInboundHandlerContext ctx) throws Exception { + in.channelActive(ctx); + } + + @Override + public void channelInactive(ChannelInboundHandlerContext ctx) throws Exception { + in.channelInactive(ctx); + } + + @Override + public void exceptionCaught( + ChannelInboundHandlerContext ctx, Throwable cause) throws Exception { + in.exceptionCaught(ctx, cause); + } + + @Override + public void userEventTriggered( + ChannelInboundHandlerContext ctx, Object evt) throws Exception { + in.userEventTriggered(ctx, evt); + } + + @Override + public void inboundBufferUpdated(ChannelInboundHandlerContext ctx) throws Exception { + in.inboundBufferUpdated(ctx); + } + + @Override + public void bind( + ChannelOutboundHandlerContext ctx, + SocketAddress localAddress, ChannelFuture future) throws Exception { + out.bind(ctx, localAddress, future); + } + + @Override + public void connect( + ChannelOutboundHandlerContext ctx, + SocketAddress remoteAddress, SocketAddress localAddress, + ChannelFuture future) throws Exception { + out.connect(ctx, remoteAddress, localAddress, future); + } + + @Override + public void disconnect( + ChannelOutboundHandlerContext ctx, ChannelFuture future) throws Exception { + out.disconnect(ctx, future); + } + + @Override + public void close( + ChannelOutboundHandlerContext ctx, ChannelFuture future) throws Exception { + out.close(ctx, future); + } + + @Override + public void deregister( + ChannelOutboundHandlerContext ctx, ChannelFuture future) throws Exception { + out.deregister(ctx, future); + } + + @Override + public void flush( + ChannelOutboundHandlerContext ctx, ChannelFuture future) throws Exception { + out.flush(ctx, future); + } + }; + } + @Override public void beforeAdd(ChannelHandlerContext ctx) throws Exception { // Do nothing by default.