andreacavalli/netty5
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

[#1812] Rework ByteToMessageDecoder.channelRead(..) method to allow for inlining

Browse Source
This commit is contained in:
Norman Maurer 2013-10-23 13:55:53 +02:00
parent 3de7a0bf76
commit 5a844d0bd6
1 changed files with 38 additions and 44 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

60
codec/src/main/java/io/netty/handler/codec/ByteToMessageDecoder.java
View File

@ -49,6 +49,7 @@ public abstract class ByteToMessageDecoder extends ChannelInboundHandlerAdapter
ByteBuf cumulation; ByteBuf cumulation;
private boolean singleDecode; private boolean singleDecode;
private boolean decodeWasNull; private boolean decodeWasNull;
private boolean first;
protected ByteToMessageDecoder() { protected ByteToMessageDecoder() {
if (getClass().isAnnotationPresent(Sharable.class)) { if (getClass().isAnnotationPresent(Sharable.class)) {
@ -121,64 +122,57 @@ public abstract class ByteToMessageDecoder extends ChannelInboundHandlerAdapter
@Override @Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
if (msg instanceof ByteBuf) {
RecyclableArrayList out = RecyclableArrayList.newInstance(); RecyclableArrayList out = RecyclableArrayList.newInstance();
try { try {
if (msg instanceof ByteBuf) {
ByteBuf data = (ByteBuf) msg; ByteBuf data = (ByteBuf) msg;
if (cumulation == null) { first = cumulation == null;
if (first) {
cumulation = data; cumulation = data;
try {
callDecode(ctx, cumulation, out);
} finally {
if (cumulation != null && !cumulation.isReadable()) {
cumulation.release();
cumulation = null;
}
}
} else { } else {
try {
if (cumulation.writerIndex() > cumulation.maxCapacity() - data.readableBytes()) { if (cumulation.writerIndex() > cumulation.maxCapacity() - data.readableBytes()) {
ByteBuf oldCumulation = cumulation; expandCumulation(ctx, data.readableBytes());
cumulation = ctx.alloc().buffer(oldCumulation.readableBytes() + data.readableBytes());
cumulation.writeBytes(oldCumulation);
oldCumulation.release();
} }
cumulation.writeBytes(data); cumulation.writeBytes(data);
callDecode(ctx, cumulation, out);
} finally {
if (cumulation != null) {
if (!cumulation.isReadable()) {
cumulation.release();
cumulation = null;
} else {
cumulation.discardSomeReadBytes();
}
}
data.release(); data.release();
} }
} callDecode(ctx, cumulation, out);
} else {
out.add(msg);
}
} catch (DecoderException e) { } catch (DecoderException e) {
throw e; throw e;
} catch (Throwable t) { } catch (Throwable t) {
throw new DecoderException(t); throw new DecoderException(t);
} finally { } finally {
if (cumulation != null && !cumulation.isReadable()) {
cumulation.release();
cumulation = null;
}
int size = out.size(); int size = out.size();
if (size == 0) { decodeWasNull = size == 0;
decodeWasNull = true;
} else {
for (int i = 0; i < size; i ++) { for (int i = 0; i < size; i ++) {
ctx.fireChannelRead(out.get(i)); ctx.fireChannelRead(out.get(i));
} }
}
out.recycle(); out.recycle();
} }
} else {
ctx.fireChannelRead(msg);
}
}
private void expandCumulation(ChannelHandlerContext ctx, int readable) {
ByteBuf oldCumulation = cumulation;
cumulation = ctx.alloc().buffer(oldCumulation.readableBytes() + readable);
cumulation.writeBytes(oldCumulation);
oldCumulation.release();
} }
@Override @Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
if (cumulation != null && !first) {
// discard some bytes if possible to make more room in the
// buffer
cumulation.discardSomeReadBytes();
}
if (decodeWasNull) { if (decodeWasNull) {
decodeWasNull = false; decodeWasNull = false;
if (!ctx.channel().config().isAutoRead()) { if (!ctx.channel().config().isAutoRead()) {