Refactor the pipeline API to support stacked codecs

- Previous API did not support the pipeline which contains multiple
  MessageToStreamEncoders because there was no way to find the closest
  outbound byte buffer.  Now you always get the correct buffer even if
  the handler that provides the buffer is placed distantly.
  For example:
  
    Channel -> MsgAEncoder -> MsgBEncoder -> MsgCEncoder
  
  Msg(A|B|C)Encoder will all have access to the channel's outbound
  byte buffer.  Previously, it was simply impossible.

- Improved ChannelBufferHolder.toString()
This commit is contained in:
Trustin Lee 2012-05-29 12:09:29 -07:00
parent 81e8c49931
commit 026715e818
35 changed files with 171 additions and 123 deletions

View File

@ -78,7 +78,7 @@ public class SpdySessionHandler extends ChannelHandlerAdapter<Object, Object> {
@Override @Override
public void inboundBufferUpdated(ChannelInboundHandlerContext<Object> ctx) public void inboundBufferUpdated(ChannelInboundHandlerContext<Object> ctx)
throws Exception { throws Exception {
Queue<Object> in = ctx.in().messageBuffer(); Queue<Object> in = ctx.inbound().messageBuffer();
for (;;) { for (;;) {
Object msg = in.poll(); Object msg = in.poll();
if (msg == null) { if (msg == null) {
@ -272,7 +272,7 @@ public class SpdySessionHandler extends ChannelHandlerAdapter<Object, Object> {
} }
} }
ctx.nextIn().messageBuffer().add(msg); ctx.nextInboundMessageBuffer().add(msg);
} }
@Override @Override
@ -303,7 +303,7 @@ public class SpdySessionHandler extends ChannelHandlerAdapter<Object, Object> {
public void flush(ChannelOutboundHandlerContext<Object> ctx, public void flush(ChannelOutboundHandlerContext<Object> ctx,
ChannelFuture future) throws Exception { ChannelFuture future) throws Exception {
Queue<Object> in = ctx.prevOut().messageBuffer(); Queue<Object> in = ctx.outbound().messageBuffer();
for (;;) { for (;;) {
Object msg = in.poll(); Object msg = in.poll();
if (msg == null) { if (msg == null) {
@ -394,7 +394,7 @@ public class SpdySessionHandler extends ChannelHandlerAdapter<Object, Object> {
} }
} }
ctx.out().messageBuffer().add(msg); ctx.nextOutboundMessageBuffer().add(msg);
} }
/* /*

View File

@ -248,7 +248,7 @@ public abstract class AbstractSocketSpdyEchoTest {
@Override @Override
public void inboundBufferUpdated(ChannelInboundHandlerContext<Byte> ctx) public void inboundBufferUpdated(ChannelInboundHandlerContext<Byte> ctx)
throws Exception { throws Exception {
ChannelBuffer m = ctx.in().byteBuffer().readBytes(ctx.in().byteBuffer().readableBytes()); ChannelBuffer m = ctx.inbound().byteBuffer().readBytes(ctx.inbound().byteBuffer().readableBytes());
byte[] actual = new byte[m.readableBytes()]; byte[] actual = new byte[m.readableBytes()];
m.getBytes(0, actual); m.getBytes(0, actual);

View File

@ -19,7 +19,7 @@ public abstract class MessageToMessageDecoder<I, O> extends ChannelInboundHandle
@Override @Override
public void inboundBufferUpdated(ChannelInboundHandlerContext<I> ctx) public void inboundBufferUpdated(ChannelInboundHandlerContext<I> ctx)
throws Exception { throws Exception {
Queue<I> in = ctx.in().messageBuffer(); Queue<I> in = ctx.inbound().messageBuffer();
boolean decoded = false; boolean decoded = false;
for (;;) { for (;;) {
try { try {
@ -35,7 +35,7 @@ public abstract class MessageToMessageDecoder<I, O> extends ChannelInboundHandle
continue; continue;
} }
if (unfoldAndAdd(ctx, ctx.nextIn(), emsg)) { if (unfoldAndAdd(ctx, ctx.nextInboundMessageBuffer(), emsg)) {
decoded = true; decoded = true;
} }
} catch (Throwable t) { } catch (Throwable t) {

View File

@ -1,6 +1,5 @@
package io.netty.handler.codec; package io.netty.handler.codec;
import io.netty.buffer.ChannelBuffer;
import io.netty.channel.ChannelBufferHolder; import io.netty.channel.ChannelBufferHolder;
import io.netty.channel.ChannelBufferHolders; import io.netty.channel.ChannelBufferHolders;
import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelFuture;
@ -20,7 +19,7 @@ public abstract class MessageToMessageEncoder<I, O> extends ChannelOutboundHandl
@Override @Override
public void flush(ChannelOutboundHandlerContext<I> ctx, ChannelFuture future) throws Exception { public void flush(ChannelOutboundHandlerContext<I> ctx, ChannelFuture future) throws Exception {
Queue<I> in = ctx.prevOut().messageBuffer(); Queue<I> in = ctx.outbound().messageBuffer();
boolean encoded = false; boolean encoded = false;
for (;;) { for (;;) {
try { try {
@ -36,7 +35,7 @@ public abstract class MessageToMessageEncoder<I, O> extends ChannelOutboundHandl
continue; continue;
} }
if (unfoldAndAdd(ctx, ctx.out(), emsg)) { if (unfoldAndAdd(ctx, ctx.nextOutboundMessageBuffer(), emsg)) {
encoded = true; encoded = true;
} }
} catch (Throwable t) { } catch (Throwable t) {
@ -56,7 +55,7 @@ public abstract class MessageToMessageEncoder<I, O> extends ChannelOutboundHandl
public abstract O encode(ChannelOutboundHandlerContext<I> ctx, I msg) throws Exception; public abstract O encode(ChannelOutboundHandlerContext<I> ctx, I msg) throws Exception;
static <T> boolean unfoldAndAdd( static <T> boolean unfoldAndAdd(
ChannelHandlerContext ctx, ChannelBufferHolder<Object> dst, Object msg) throws Exception { ChannelHandlerContext ctx, Queue<Object> dst, Object msg) throws Exception {
if (msg == null) { if (msg == null) {
return false; return false;
} }
@ -94,18 +93,7 @@ public abstract class MessageToMessageEncoder<I, O> extends ChannelOutboundHandl
return added; return added;
} }
if (dst.hasMessageBuffer()) { dst.add(msg);
dst.messageBuffer().add(msg);
} else if (msg instanceof ChannelBuffer) {
ChannelBuffer buf = (ChannelBuffer) msg;
if (!buf.readable()) {
return false;
}
dst.byteBuffer().writeBytes(buf, buf.readerIndex(), buf.readableBytes());
} else {
throw new UnsupportedMessageTypeException(msg, ChannelBuffer.class);
}
return true; return true;
} }
} }

View File

@ -19,8 +19,8 @@ public abstract class MessageToStreamEncoder<I> extends ChannelOutboundHandlerAd
@Override @Override
public void flush(ChannelOutboundHandlerContext<I> ctx, ChannelFuture future) throws Exception { public void flush(ChannelOutboundHandlerContext<I> ctx, ChannelFuture future) throws Exception {
Queue<I> in = ctx.prevOut().messageBuffer(); Queue<I> in = ctx.outbound().messageBuffer();
ChannelBuffer out = ctx.out().byteBuffer(); ChannelBuffer out = ctx.nextOutboundByteBuffer();
int oldOutSize = out.readableBytes(); int oldOutSize = out.readableBytes();
for (;;) { for (;;) {

View File

@ -378,7 +378,7 @@ public abstract class ReplayingDecoder<O, S extends Enum<S>> extends StreamToMes
} }
try { try {
if (unfoldAndAdd(ctx, ctx.nextIn(), decodeLast(ctx, replayable))) { if (unfoldAndAdd(ctx, ctx.nextInboundMessageBuffer(), decodeLast(ctx, replayable))) {
fireInboundBufferUpdated(ctx, in); fireInboundBufferUpdated(ctx, in);
} }
} catch (Signal replay) { } catch (Signal replay) {
@ -442,7 +442,7 @@ public abstract class ReplayingDecoder<O, S extends Enum<S>> extends StreamToMes
} }
// A successful decode // A successful decode
if (unfoldAndAdd(ctx, ctx.nextIn(), result)) { if (unfoldAndAdd(ctx, ctx.nextInboundMessageBuffer(), result)) {
decoded = true; decoded = true;
} }
} catch (Throwable t) { } catch (Throwable t) {

View File

@ -27,13 +27,13 @@ public abstract class StreamToMessageDecoder<O> extends ChannelInboundHandlerAda
@Override @Override
public void channelInactive(ChannelInboundHandlerContext<Byte> ctx) throws Exception { public void channelInactive(ChannelInboundHandlerContext<Byte> ctx) throws Exception {
ChannelBuffer in = ctx.in().byteBuffer(); ChannelBuffer in = ctx.inbound().byteBuffer();
if (in.readable()) { if (in.readable()) {
callDecode(ctx); callDecode(ctx);
} }
try { try {
if (unfoldAndAdd(ctx, ctx.nextIn(), decodeLast(ctx, in))) { if (unfoldAndAdd(ctx, ctx.nextInboundMessageBuffer(), decodeLast(ctx, in))) {
in.discardReadBytes(); in.discardReadBytes();
ctx.fireInboundBufferUpdated(); ctx.fireInboundBufferUpdated();
} }
@ -49,7 +49,7 @@ public abstract class StreamToMessageDecoder<O> extends ChannelInboundHandlerAda
} }
protected void callDecode(ChannelInboundHandlerContext<Byte> ctx) { protected void callDecode(ChannelInboundHandlerContext<Byte> ctx) {
ChannelBuffer in = ctx.in().byteBuffer(); ChannelBuffer in = ctx.inbound().byteBuffer();
boolean decoded = false; boolean decoded = false;
for (;;) { for (;;) {
@ -69,7 +69,7 @@ public abstract class StreamToMessageDecoder<O> extends ChannelInboundHandlerAda
} }
} }
if (unfoldAndAdd(ctx, ctx.nextIn(), o)) { if (unfoldAndAdd(ctx, ctx.nextInboundMessageBuffer(), o)) {
decoded = true; decoded = true;
} else { } else {
break; break;
@ -109,10 +109,10 @@ public abstract class StreamToMessageDecoder<O> extends ChannelInboundHandlerAda
// the new handler. // the new handler.
ctx.pipeline().addAfter(ctx.name(), newHandlerName, newHandler); ctx.pipeline().addAfter(ctx.name(), newHandlerName, newHandler);
ChannelBuffer in = ctx.in().byteBuffer(); ChannelBuffer in = ctx.inbound().byteBuffer();
try { try {
if (in.readable()) { if (in.readable()) {
ctx.nextIn().byteBuffer().writeBytes(ctx.in().byteBuffer()); ctx.nextInboundByteBuffer().writeBytes(ctx.inbound().byteBuffer());
ctx.fireInboundBufferUpdated(); ctx.fireInboundBufferUpdated();
} }
} finally { } finally {

View File

@ -21,12 +21,12 @@ public abstract class StreamToStreamDecoder extends ChannelInboundHandlerAdapter
@Override @Override
public void channelInactive(ChannelInboundHandlerContext<Byte> ctx) throws Exception { public void channelInactive(ChannelInboundHandlerContext<Byte> ctx) throws Exception {
ChannelBuffer in = ctx.in().byteBuffer(); ChannelBuffer in = ctx.inbound().byteBuffer();
if (!in.readable()) { if (!in.readable()) {
callDecode(ctx); callDecode(ctx);
} }
ChannelBuffer out = ctx.nextIn().byteBuffer(); ChannelBuffer out = ctx.nextInboundByteBuffer();
int oldOutSize = out.readableBytes(); int oldOutSize = out.readableBytes();
try { try {
decodeLast(ctx, in, out); decodeLast(ctx, in, out);
@ -47,8 +47,8 @@ public abstract class StreamToStreamDecoder extends ChannelInboundHandlerAdapter
} }
private void callDecode(ChannelInboundHandlerContext<Byte> ctx) { private void callDecode(ChannelInboundHandlerContext<Byte> ctx) {
ChannelBuffer in = ctx.in().byteBuffer(); ChannelBuffer in = ctx.inbound().byteBuffer();
ChannelBuffer out = ctx.nextIn().byteBuffer(); ChannelBuffer out = ctx.nextInboundByteBuffer();
int oldOutSize = out.readableBytes(); int oldOutSize = out.readableBytes();
while (in.readable()) { while (in.readable()) {

View File

@ -17,8 +17,8 @@ public abstract class StreamToStreamEncoder extends ChannelOutboundHandlerAdapte
@Override @Override
public void flush(ChannelOutboundHandlerContext<Byte> ctx, ChannelFuture future) throws Exception { public void flush(ChannelOutboundHandlerContext<Byte> ctx, ChannelFuture future) throws Exception {
ChannelBuffer in = ctx.prevOut().byteBuffer(); ChannelBuffer in = ctx.outbound().byteBuffer();
ChannelBuffer out = ctx.out().byteBuffer(); ChannelBuffer out = ctx.nextOutboundByteBuffer();
int oldOutSize = out.readableBytes(); int oldOutSize = out.readableBytes();
while (in.readable()) { while (in.readable()) {

View File

@ -57,7 +57,7 @@ public class DecoderEmbedder<E> extends AbstractCodecEmbedder<E> {
@Override @Override
public boolean offer(Object input) { public boolean offer(Object input) {
ChannelBufferHolder<Object> in = pipeline().nextIn(); ChannelBufferHolder<Object> in = pipeline().inbound();
if (in.hasByteBuffer()) { if (in.hasByteBuffer()) {
in.byteBuffer().writeBytes((ChannelBuffer) input); in.byteBuffer().writeBytes((ChannelBuffer) input);
} else { } else {

View File

@ -34,7 +34,6 @@ public class DiscardClientHandler extends ChannelInboundStreamHandlerAdapter {
private final byte[] content; private final byte[] content;
private ChannelInboundHandlerContext<Byte> ctx; private ChannelInboundHandlerContext<Byte> ctx;
private ChannelBuffer out;
public DiscardClientHandler(int messageSize) { public DiscardClientHandler(int messageSize) {
if (messageSize <= 0) { if (messageSize <= 0) {
@ -49,7 +48,6 @@ public class DiscardClientHandler extends ChannelInboundStreamHandlerAdapter {
public void channelActive(ChannelInboundHandlerContext<Byte> ctx) public void channelActive(ChannelInboundHandlerContext<Byte> ctx)
throws Exception { throws Exception {
this.ctx = ctx; this.ctx = ctx;
out = ctx.out().byteBuffer();
// Send the initial messages. // Send the initial messages.
generateTraffic(); generateTraffic();
} }
@ -77,6 +75,7 @@ public class DiscardClientHandler extends ChannelInboundStreamHandlerAdapter {
private void generateTraffic() { private void generateTraffic() {
// Fill the outbound buffer up to 64KiB // Fill the outbound buffer up to 64KiB
ChannelBuffer out = ctx.nextOutboundByteBuffer();
while (out.readableBytes() < 65536) { while (out.readableBytes() < 65536) {
out.writeBytes(content); out.writeBytes(content);
} }
@ -90,7 +89,7 @@ public class DiscardClientHandler extends ChannelInboundStreamHandlerAdapter {
@Override @Override
public void operationComplete(ChannelFuture future) throws Exception { public void operationComplete(ChannelFuture future) throws Exception {
if (future.isSuccess()) { if (future.isSuccess()) {
out.clear(); ctx.nextOutboundByteBuffer().discardReadBytes();
generateTraffic(); generateTraffic();
} }
} }

View File

@ -34,7 +34,7 @@ public class DiscardServerHandler extends ChannelInboundStreamHandlerAdapter {
public void inboundBufferUpdated(ChannelInboundHandlerContext<Byte> ctx) public void inboundBufferUpdated(ChannelInboundHandlerContext<Byte> ctx)
throws Exception { throws Exception {
// Discard the received data silently. // Discard the received data silently.
ctx.in().byteBuffer().clear(); ctx.inbound().byteBuffer().clear();
} }

View File

@ -62,8 +62,8 @@ public class EchoClientHandler extends ChannelInboundHandlerAdapter<Byte> {
@Override @Override
public void inboundBufferUpdated(ChannelInboundHandlerContext<Byte> ctx) { public void inboundBufferUpdated(ChannelInboundHandlerContext<Byte> ctx) {
ChannelBuffer in = ctx.in().byteBuffer(); ChannelBuffer in = ctx.inbound().byteBuffer();
ChannelBuffer out = ctx.out().byteBuffer(); ChannelBuffer out = ctx.nextOutboundByteBuffer();
out.discardReadBytes(); out.discardReadBytes();
out.writeBytes(in); out.writeBytes(in);
in.discardReadBytes(); in.discardReadBytes();

View File

@ -39,8 +39,8 @@ public class EchoServerHandler extends ChannelInboundHandlerAdapter<Byte> {
@Override @Override
public void inboundBufferUpdated(ChannelInboundHandlerContext<Byte> ctx) { public void inboundBufferUpdated(ChannelInboundHandlerContext<Byte> ctx) {
ChannelBuffer in = ctx.in().byteBuffer(); ChannelBuffer in = ctx.inbound().byteBuffer();
ChannelBuffer out = ctx.out().byteBuffer(); ChannelBuffer out = ctx.nextOutboundByteBuffer();
out.discardReadBytes(); out.discardReadBytes();
out.writeBytes(in); out.writeBytes(in);
in.discardReadBytes(); in.discardReadBytes();

View File

@ -40,7 +40,6 @@ public class FactorialClientHandler extends ChannelInboundMessageHandlerAdapter<
FactorialClientHandler.class.getName()); FactorialClientHandler.class.getName());
private ChannelInboundHandlerContext<BigInteger> ctx; private ChannelInboundHandlerContext<BigInteger> ctx;
private Queue<Object> out;
private int i = 1; private int i = 1;
private int receivedMessages; private int receivedMessages;
private final int count; private final int count;
@ -68,7 +67,6 @@ public class FactorialClientHandler extends ChannelInboundMessageHandlerAdapter<
@Override @Override
public void channelActive(ChannelInboundHandlerContext<BigInteger> ctx) { public void channelActive(ChannelInboundHandlerContext<BigInteger> ctx) {
this.ctx = ctx; this.ctx = ctx;
out = ctx.out().messageBuffer();
sendNumbers(); sendNumbers();
} }
@ -101,6 +99,7 @@ public class FactorialClientHandler extends ChannelInboundMessageHandlerAdapter<
private void sendNumbers() { private void sendNumbers() {
// Do not send more than 4096 numbers. // Do not send more than 4096 numbers.
boolean finished = false; boolean finished = false;
Queue<Object> out = ctx.nextOutboundMessageBuffer();
while (out.size() < 4096) { while (out.size() < 4096) {
if (i <= count) { if (i <= count) {
out.add(Integer.valueOf(i)); out.add(Integer.valueOf(i));

View File

@ -352,7 +352,7 @@ public class LoggingHandler extends ChannelHandlerAdapter<Object, Object> {
public void inboundBufferUpdated(ChannelInboundHandlerContext<Object> ctx) public void inboundBufferUpdated(ChannelInboundHandlerContext<Object> ctx)
throws Exception { throws Exception {
if (getLogger().isEnabled(internalLevel)) { if (getLogger().isEnabled(internalLevel)) {
logger.log(internalLevel, format(ctx, formatBuffer("INBUF", ctx.in()))); logger.log(internalLevel, format(ctx, formatBuffer("INBUF", ctx.inbound())));
} }
ctx.fireInboundBufferUpdated(); ctx.fireInboundBufferUpdated();
} }
@ -407,7 +407,7 @@ public class LoggingHandler extends ChannelHandlerAdapter<Object, Object> {
public void flush(ChannelOutboundHandlerContext<Object> ctx, public void flush(ChannelOutboundHandlerContext<Object> ctx,
ChannelFuture future) throws Exception { ChannelFuture future) throws Exception {
if (getLogger().isEnabled(internalLevel)) { if (getLogger().isEnabled(internalLevel)) {
logger.log(internalLevel, format(ctx, formatBuffer("OUTBUF", ctx.prevOut()))); logger.log(internalLevel, format(ctx, formatBuffer("OUTBUF", ctx.outbound())));
} }
ctx.flush(future); ctx.flush(future);
} }

View File

@ -259,8 +259,8 @@ public abstract class AbstractChannel extends DefaultAttributeMap implements Cha
} }
@Override @Override
public ChannelBufferHolder<Object> out() { public ChannelBufferHolder<Object> outbound() {
return pipeline().out(); return pipeline().outbound();
} }
@Override @Override

View File

@ -37,7 +37,7 @@ public abstract class AbstractServerChannel extends AbstractChannel implements S
} }
@Override @Override
public ChannelBufferHolder<Object> out() { public ChannelBufferHolder<Object> outbound() {
return ChannelBufferHolders.discardBuffer(); return ChannelBufferHolders.discardBuffer();
} }

View File

@ -136,6 +136,8 @@ public interface Channel extends AttributeMap, ChannelOutboundInvoker, ChannelFu
boolean isRegistered(); boolean isRegistered();
boolean isActive(); boolean isActive();
ChannelBufferHolder<Object> outbound();
/** /**
* Returns the local address where this channel is bound to. The returned * Returns the local address where this channel is bound to. The returned
* {@link SocketAddress} is supposed to be down-cast into more concrete * {@link SocketAddress} is supposed to be down-cast into more concrete

View File

@ -59,9 +59,9 @@ public final class ChannelBufferHolder<E> {
case 0: case 0:
return msgBuf != null; return msgBuf != null;
case 1: case 1:
return ctx.nextIn().hasMessageBuffer(); return ctx.nextInboundMessageBuffer() != null;
case 2: case 2:
return ctx.out().hasMessageBuffer(); return ctx.nextOutboundMessageBuffer() != null;
default: default:
throw new Error(); throw new Error();
} }
@ -72,15 +72,14 @@ public final class ChannelBufferHolder<E> {
case 0: case 0:
return byteBuf != null; return byteBuf != null;
case 1: case 1:
return ctx.nextIn().hasByteBuffer(); return ctx.nextInboundByteBuffer() != null;
case 2: case 2:
return ctx.out().hasByteBuffer(); return ctx.nextOutboundByteBuffer() != null;
default: default:
throw new Error(); throw new Error();
} }
} }
@SuppressWarnings("unchecked")
public Queue<E> messageBuffer() { public Queue<E> messageBuffer() {
switch (bypassDirection) { switch (bypassDirection) {
case 0: case 0:
@ -89,9 +88,9 @@ public final class ChannelBufferHolder<E> {
} }
return msgBuf; return msgBuf;
case 1: case 1:
return (Queue<E>) ctx.nextIn().messageBuffer(); return (Queue<E>) ctx.nextInboundMessageBuffer();
case 2: case 2:
return (Queue<E>) ctx.out().messageBuffer(); return (Queue<E>) ctx.nextOutboundMessageBuffer();
default: default:
throw new Error(); throw new Error();
} }
@ -105,9 +104,9 @@ public final class ChannelBufferHolder<E> {
} }
return byteBuf; return byteBuf;
case 1: case 1:
return ctx.nextIn().byteBuffer(); return ctx.nextInboundByteBuffer();
case 2: case 2:
return ctx.out().byteBuffer(); return ctx.nextOutboundByteBuffer();
default: default:
throw new Error(); throw new Error();
} }
@ -118,14 +117,18 @@ public final class ChannelBufferHolder<E> {
switch (bypassDirection) { switch (bypassDirection) {
case 0: case 0:
if (msgBuf != null) { if (msgBuf != null) {
return msgBuf.toString(); if (byteBuf != null) {
return "CatchAllBuffer";
} else {
return "MessageBuffer(" + msgBuf.size() + ')';
}
} else { } else {
return byteBuf.toString(); return byteBuf.toString();
} }
case 1: case 1:
return ctx.nextIn().toString(); return "InboundBypassBuffer";
case 2: case 2:
return ctx.out().toString(); return "OutboundBypassBuffer";
default: default:
throw new Error(); throw new Error();
} }
@ -140,9 +143,8 @@ public final class ChannelBufferHolder<E> {
return byteBuf.readableBytes(); return byteBuf.readableBytes();
} }
case 1: case 1:
return ctx.nextIn().size();
case 2: case 2:
return ctx.out().size(); throw new UnsupportedOperationException();
default: default:
throw new Error(); throw new Error();
} }
@ -157,9 +159,8 @@ public final class ChannelBufferHolder<E> {
return byteBuf.readable(); return byteBuf.readable();
} }
case 1: case 1:
return ctx.nextIn().isEmpty();
case 2: case 2:
return ctx.out().isEmpty(); throw new UnsupportedOperationException();
default: default:
throw new Error(); throw new Error();
} }

View File

@ -16,9 +16,11 @@
package io.netty.channel; package io.netty.channel;
import io.netty.buffer.ChannelBuffer;
import io.netty.util.AttributeMap; import io.netty.util.AttributeMap;
import java.nio.channels.Channels; import java.nio.channels.Channels;
import java.util.Queue;
/** /**
* Enables a {@link ChannelHandler} to interact with its {@link ChannelPipeline} * Enables a {@link ChannelHandler} to interact with its {@link ChannelPipeline}
@ -133,4 +135,10 @@ public interface ChannelHandlerContext
boolean canHandleInbound(); boolean canHandleInbound();
boolean canHandleOutbound(); boolean canHandleOutbound();
ChannelBuffer nextInboundByteBuffer();
Queue<Object> nextInboundMessageBuffer();
ChannelBuffer nextOutboundByteBuffer();
Queue<Object> nextOutboundMessageBuffer();
} }

View File

@ -61,14 +61,14 @@ public abstract class ChannelInboundHandlerAdapter<I> implements ChannelInboundH
} }
static <I> void inboundBufferUpdated0(ChannelInboundHandlerContext<I> ctx) { static <I> void inboundBufferUpdated0(ChannelInboundHandlerContext<I> ctx) {
if (ctx.in().isBypass()) { if (ctx.inbound().isBypass()) {
ctx.fireInboundBufferUpdated(); ctx.fireInboundBufferUpdated();
return; return;
} }
if (ctx.in().hasMessageBuffer()) { if (ctx.inbound().hasMessageBuffer()) {
Queue<I> in = ctx.in().messageBuffer(); Queue<I> in = ctx.inbound().messageBuffer();
Queue<Object> nextIn = ctx.nextIn().messageBuffer(); Queue<Object> nextIn = ctx.nextInboundMessageBuffer();
for (;;) { for (;;) {
I msg = in.poll(); I msg = in.poll();
if (msg == null) { if (msg == null) {
@ -77,8 +77,8 @@ public abstract class ChannelInboundHandlerAdapter<I> implements ChannelInboundH
nextIn.add(msg); nextIn.add(msg);
} }
} else { } else {
ChannelBuffer in = ctx.in().byteBuffer(); ChannelBuffer in = ctx.inbound().byteBuffer();
ChannelBuffer nextIn = ctx.nextIn().byteBuffer(); ChannelBuffer nextIn = ctx.nextInboundByteBuffer();
nextIn.writeBytes(in); nextIn.writeBytes(in);
in.discardReadBytes(); in.discardReadBytes();
} }

View File

@ -1,5 +1,5 @@
package io.netty.channel; package io.netty.channel;
public interface ChannelInboundHandlerContext<I> extends ChannelHandlerContext { public interface ChannelInboundHandlerContext<I> extends ChannelHandlerContext {
ChannelBufferHolder<I> in(); ChannelBufferHolder<I> inbound();
} }

View File

@ -2,8 +2,6 @@ package io.netty.channel;
public interface ChannelInboundInvoker { public interface ChannelInboundInvoker {
ChannelBufferHolder<Object> nextIn();
void fireChannelRegistered(); void fireChannelRegistered();
void fireChannelUnregistered(); void fireChannelUnregistered();
void fireChannelActive(); void fireChannelActive();

View File

@ -14,7 +14,7 @@ public class ChannelInboundMessageHandlerAdapter<I> extends
@Override @Override
public void inboundBufferUpdated(ChannelInboundHandlerContext<I> ctx) public void inboundBufferUpdated(ChannelInboundHandlerContext<I> ctx)
throws Exception { throws Exception {
Queue<I> in = ctx.in().messageBuffer(); Queue<I> in = ctx.inbound().messageBuffer();
for (;;) { for (;;) {
I msg = in.poll(); I msg = in.poll();
if (msg == null) { if (msg == null) {
@ -30,6 +30,6 @@ public class ChannelInboundMessageHandlerAdapter<I> extends
} }
public void messageReceived(ChannelInboundHandlerContext<I> ctx, I msg) throws Exception { public void messageReceived(ChannelInboundHandlerContext<I> ctx, I msg) throws Exception {
ctx.nextIn().messageBuffer().add(msg); ctx.nextInboundMessageBuffer().add(msg);
} }
} }

View File

@ -57,14 +57,14 @@ public abstract class ChannelOutboundHandlerAdapter<O> implements ChannelOutboun
} }
static <O> void flush0(ChannelOutboundHandlerContext<O> ctx, ChannelFuture future) { static <O> void flush0(ChannelOutboundHandlerContext<O> ctx, ChannelFuture future) {
if (ctx.prevOut().isBypass()) { if (ctx.outbound().isBypass()) {
ctx.flush(future); ctx.flush(future);
return; return;
} }
if (ctx.prevOut().hasMessageBuffer()) { if (ctx.outbound().hasMessageBuffer()) {
Queue<O> out = ctx.prevOut().messageBuffer(); Queue<O> out = ctx.outbound().messageBuffer();
Queue<Object> nextOut = ctx.out().messageBuffer(); Queue<Object> nextOut = ctx.nextOutboundMessageBuffer();
for (;;) { for (;;) {
O msg = out.poll(); O msg = out.poll();
if (msg == null) { if (msg == null) {
@ -73,8 +73,8 @@ public abstract class ChannelOutboundHandlerAdapter<O> implements ChannelOutboun
nextOut.add(msg); nextOut.add(msg);
} }
} else { } else {
ChannelBuffer out = ctx.prevOut().byteBuffer(); ChannelBuffer out = ctx.outbound().byteBuffer();
ChannelBuffer nextOut = ctx.out().byteBuffer(); ChannelBuffer nextOut = ctx.nextOutboundByteBuffer();
nextOut.writeBytes(out); nextOut.writeBytes(out);
out.discardReadBytes(); out.discardReadBytes();
} }

View File

@ -2,5 +2,5 @@ package io.netty.channel;
public interface ChannelOutboundHandlerContext<O> extends ChannelHandlerContext { public interface ChannelOutboundHandlerContext<O> extends ChannelHandlerContext {
ChannelBufferHolder<O> prevOut(); ChannelBufferHolder<O> outbound();
} }

View File

@ -3,8 +3,6 @@ package io.netty.channel;
import java.net.SocketAddress; import java.net.SocketAddress;
public interface ChannelOutboundInvoker { public interface ChannelOutboundInvoker {
ChannelBufferHolder<Object> out();
ChannelFuture bind(SocketAddress localAddress); ChannelFuture bind(SocketAddress localAddress);
ChannelFuture connect(SocketAddress remoteAddress); ChannelFuture connect(SocketAddress remoteAddress);
ChannelFuture connect(SocketAddress remoteAddress, SocketAddress localAddress); ChannelFuture connect(SocketAddress remoteAddress, SocketAddress localAddress);

View File

@ -205,6 +205,9 @@ import java.util.NoSuchElementException;
*/ */
public interface ChannelPipeline extends ChannelInboundInvoker, ChannelOutboundInvoker { public interface ChannelPipeline extends ChannelInboundInvoker, ChannelOutboundInvoker {
ChannelBufferHolder<Object> inbound();
ChannelBufferHolder<Object> outbound();
/** /**
* Inserts a {@link ChannelHandler} at the first position of this pipeline. * Inserts a {@link ChannelHandler} at the first position of this pipeline.
* *

View File

@ -27,6 +27,7 @@ import java.util.LinkedHashMap;
import java.util.List; import java.util.List;
import java.util.Map; import java.util.Map;
import java.util.NoSuchElementException; import java.util.NoSuchElementException;
import java.util.Queue;
/** /**
* The default {@link ChannelPipeline} implementation. It is usually created * The default {@link ChannelPipeline} implementation. It is usually created
@ -570,19 +571,19 @@ public class DefaultChannelPipeline implements ChannelPipeline {
} }
@Override @Override
public ChannelBufferHolder<Object> nextIn() { public ChannelBufferHolder<Object> inbound() {
DefaultChannelHandlerContext ctx = firstInboundContext(); DefaultChannelHandlerContext ctx = firstInboundContext();
if (ctx != null) { if (ctx != null) {
return ctx.in(); return ctx.inbound();
} }
return null; return null;
} }
@Override @Override
public ChannelBufferHolder<Object> out() { public ChannelBufferHolder<Object> outbound() {
DefaultChannelHandlerContext ctx = firstOutboundContext(); DefaultChannelHandlerContext ctx = firstOutboundContext();
if (ctx != null) { if (ctx != null) {
return ctx.prevOut(); return ctx.outbound();
} }
return channel().unsafe().out(); return channel().unsafe().out();
} }
@ -903,11 +904,12 @@ public class DefaultChannelPipeline implements ChannelPipeline {
} }
validateFuture(future); validateFuture(future);
if (out().hasMessageBuffer()) { ChannelBufferHolder<Object> out = outbound();
out().messageBuffer().add(message); if (out.hasMessageBuffer()) {
out.messageBuffer().add(message);
} else if (message instanceof ChannelBuffer) { } else if (message instanceof ChannelBuffer) {
ChannelBuffer m = (ChannelBuffer) message; ChannelBuffer m = (ChannelBuffer) message;
out().byteBuffer().writeBytes(m, m.readerIndex(), m.readableBytes()); out.byteBuffer().writeBytes(m, m.readerIndex(), m.readableBytes());
} else { } else {
throw new IllegalArgumentException( throw new IllegalArgumentException(
"cannot write a message whose type is not " + "cannot write a message whose type is not " +
@ -1057,7 +1059,7 @@ public class DefaultChannelPipeline implements ChannelPipeline {
private final boolean canHandleInbound; private final boolean canHandleInbound;
private final boolean canHandleOutbound; private final boolean canHandleOutbound;
private final ChannelBufferHolder<Object> in; private final ChannelBufferHolder<Object> in;
private final ChannelBufferHolder<Object> prevOut; private final ChannelBufferHolder<Object> out;
@SuppressWarnings("unchecked") @SuppressWarnings("unchecked")
DefaultChannelHandlerContext( DefaultChannelHandlerContext(
@ -1096,7 +1098,7 @@ public class DefaultChannelPipeline implements ChannelPipeline {
} }
if (canHandleOutbound) { if (canHandleOutbound) {
try { try {
prevOut = ((ChannelOutboundHandler<Object>) handler).newOutboundBuffer(this); out = ((ChannelOutboundHandler<Object>) handler).newOutboundBuffer(this);
} catch (Exception e) { } catch (Exception e) {
throw new ChannelPipelineException("A user handler failed to create a new outbound buffer.", e); throw new ChannelPipelineException("A user handler failed to create a new outbound buffer.", e);
} finally { } finally {
@ -1105,7 +1107,7 @@ public class DefaultChannelPipeline implements ChannelPipeline {
} }
} }
} else { } else {
prevOut = null; out = null;
} }
} }
@ -1140,32 +1142,82 @@ public class DefaultChannelPipeline implements ChannelPipeline {
} }
@Override @Override
public ChannelBufferHolder<Object> in() { public ChannelBufferHolder<Object> inbound() {
return in; return in;
} }
@Override @Override
public ChannelBufferHolder<Object> prevOut() { public ChannelBufferHolder<Object> outbound() {
return prevOut; return out;
} }
@Override @Override
public ChannelBufferHolder<Object> nextIn() { public ChannelBuffer nextInboundByteBuffer() {
DefaultChannelHandlerContext next = nextInboundContext(this.next); DefaultChannelHandlerContext ctx = this;
if (next != null) { for (;;) {
return next.in(); ctx = nextInboundContext(ctx.next);
} else { if (ctx == null) {
throw new NoSuchElementException("no inbound buffer in the rest of the pipeline"); return null;
}
ChannelBufferHolder<Object> nextIn = ctx.inbound();
if (nextIn.hasByteBuffer()) {
return nextIn.byteBuffer();
}
} }
} }
@Override @Override
public ChannelBufferHolder<Object> out() { public Queue<Object> nextInboundMessageBuffer() {
DefaultChannelHandlerContext next = nextOutboundContext(prev); DefaultChannelHandlerContext ctx = this;
if (next != null) { for (;;) {
return next.prevOut(); ctx = nextInboundContext(ctx.next);
} else { if (ctx == null) {
return channel().unsafe().out(); return null;
}
ChannelBufferHolder<Object> nextIn = ctx.inbound();
if (nextIn.hasMessageBuffer()) {
return nextIn.messageBuffer();
}
}
}
@Override
public ChannelBuffer nextOutboundByteBuffer() {
DefaultChannelHandlerContext ctx = this;
for (;;) {
ctx = nextOutboundContext(ctx.prev);
if (ctx == null) {
ChannelBufferHolder<Object> lastOut = channel().unsafe().out();
if (lastOut.hasByteBuffer()) {
return lastOut.byteBuffer();
} else {
return null;
}
}
ChannelBufferHolder<Object> nextOut = ctx.outbound();
if (nextOut.hasByteBuffer()) {
return nextOut.byteBuffer();
}
}
}
@Override
public Queue<Object> nextOutboundMessageBuffer() {
DefaultChannelHandlerContext ctx = this;
for (;;) {
ctx = nextOutboundContext(ctx.prev);
if (ctx == null) {
ChannelBufferHolder<Object> lastOut = channel().unsafe().out();
if (lastOut.hasMessageBuffer()) {
return lastOut.messageBuffer();
} else {
return null;
}
}
ChannelBufferHolder<Object> nextOut = ctx.outbound();
if (nextOut.hasMessageBuffer()) {
return nextOut.messageBuffer();
}
} }
} }
@ -1304,9 +1356,9 @@ public class DefaultChannelPipeline implements ChannelPipeline {
public ChannelFuture write(Object message, ChannelFuture future) { public ChannelFuture write(Object message, ChannelFuture future) {
if (message instanceof ChannelBuffer) { if (message instanceof ChannelBuffer) {
ChannelBuffer m = (ChannelBuffer) message; ChannelBuffer m = (ChannelBuffer) message;
out().byteBuffer().writeBytes(m, m.readerIndex(), m.readableBytes()); nextOutboundByteBuffer().writeBytes(m, m.readerIndex(), m.readableBytes());
} else { } else {
out().messageBuffer().add(message); nextOutboundMessageBuffer().add(message);
} }
return flush(future); return flush(future);
} }

View File

@ -180,7 +180,7 @@ public class ServerChannelBootstrap {
@Override @Override
public void inboundBufferUpdated(ChannelInboundHandlerContext<Channel> ctx) { public void inboundBufferUpdated(ChannelInboundHandlerContext<Channel> ctx) {
Queue<Channel> in = ctx.in().messageBuffer(); Queue<Channel> in = ctx.inbound().messageBuffer();
for (;;) { for (;;) {
Channel child = in.poll(); Channel child = in.poll();
if (child == null) { if (child == null) {

View File

@ -34,7 +34,7 @@ abstract class AbstractNioMessageChannel extends AbstractNioChannel {
assert eventLoop().inEventLoop(); assert eventLoop().inEventLoop();
final ChannelPipeline pipeline = pipeline(); final ChannelPipeline pipeline = pipeline();
final ChannelBufferHolder<Object> buf = pipeline.nextIn(); final ChannelBufferHolder<Object> buf = pipeline.inbound();
boolean closed = false; boolean closed = false;
boolean read = false; boolean read = false;
try { try {

View File

@ -36,7 +36,7 @@ abstract class AbstractNioStreamChannel extends AbstractNioChannel {
assert eventLoop().inEventLoop(); assert eventLoop().inEventLoop();
final ChannelPipeline pipeline = pipeline(); final ChannelPipeline pipeline = pipeline();
final ChannelBufferHolder<Object> buf = pipeline.nextIn(); final ChannelBufferHolder<Object> buf = pipeline.inbound();
boolean closed = false; boolean closed = false;
boolean read = false; boolean read = false;
try { try {

View File

@ -32,7 +32,7 @@ abstract class AbstractOioMessageChannel extends AbstractOioChannel {
assert eventLoop().inEventLoop(); assert eventLoop().inEventLoop();
final ChannelPipeline pipeline = pipeline(); final ChannelPipeline pipeline = pipeline();
final ChannelBufferHolder<Object> buf = pipeline.nextIn(); final ChannelBufferHolder<Object> buf = pipeline.inbound();
boolean closed = false; boolean closed = false;
boolean read = false; boolean read = false;
try { try {

View File

@ -33,7 +33,7 @@ abstract class AbstractOioStreamChannel extends AbstractOioChannel {
assert eventLoop().inEventLoop(); assert eventLoop().inEventLoop();
final ChannelPipeline pipeline = pipeline(); final ChannelPipeline pipeline = pipeline();
final ChannelBufferHolder<Object> buf = pipeline.nextIn(); final ChannelBufferHolder<Object> buf = pipeline.inbound();
boolean closed = false; boolean closed = false;
boolean read = false; boolean read = false;
try { try {