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

Made sure to clean up the cumulative buffer on channelDisconnected or channelClosed

Browse Source
This commit is contained in:
Trustin Lee 2009-06-04 08:49:33 +00:00
parent d4071e87ef
commit 345a5512ab
2 changed files with 27 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
src/main/java/org/jboss/netty/handler/codec/frame/FrameDecoder.java
View File

@ -293,8 +293,12 @@ public abstract class FrameDecoder extends SimpleChannelUpstreamHandler {
private void cleanup(ChannelHandlerContext ctx, ChannelStateEvent e)
throws Exception {
ChannelBuffer cumulation = cumulation(ctx);
try {
ChannelBuffer cumulation = this.cumulation.getAndSet(null);
if (cumulation == null) {
return;
}
if (cumulation.readable()) {
// Make sure all frames are read before notifying a closed channel.
callDecode(ctx, ctx.getChannel(), cumulation, null);

38
src/main/java/org/jboss/netty/handler/codec/replay/ReplayingDecoder.java
View File

@ -291,7 +291,12 @@ public abstract class ReplayingDecoder<T extends Enum<T>>
* Stores the internal cumulative buffer's reader position.
*/
protected void checkpoint() {
checkpoint = cumulation().readerIndex();
ChannelBuffer cumulation = this.cumulation.get();
if (cumulation != null) {
checkpoint = cumulation.readerIndex();
} else {
checkpoint = -1; // buffer not available (already cleaned up)
}
}
/**
@ -299,8 +304,8 @@ public abstract class ReplayingDecoder<T extends Enum<T>>
* the current decoder state.
*/
protected void checkpoint(T state) {
checkpoint = cumulation().readerIndex();
this.state = state;
checkpoint();
setState(state);
}
/**
@ -414,7 +419,13 @@ public abstract class ReplayingDecoder<T extends Enum<T>>
}
} catch (ReplayError replay) {
// Return to the checkpoint (or oldPosition) and retry.
cumulation.readerIndex(checkpoint);
int checkpoint = this.checkpoint;
if (checkpoint >= 0) {
cumulation.readerIndex(checkpoint);
} else {
// Called by cleanup() - no need to maintain the readerIndex
// anymore because the buffer has been released already.
}
}
if (result == null) {
@ -460,9 +471,14 @@ public abstract class ReplayingDecoder<T extends Enum<T>>
private void cleanup(ChannelHandlerContext ctx, ChannelStateEvent e)
throws Exception {
ChannelBuffer cumulation = cumulation(ctx);
replayable.terminate();
try {
ChannelBuffer cumulation = this.cumulation.getAndSet(null);
if (cumulation == null) {
return;
}
replayable.terminate();
if (cumulation.readable()) {
// Make sure all data was read before notifying a closed channel.
callDecode(ctx, e.getChannel(), cumulation, null);
@ -482,7 +498,6 @@ public abstract class ReplayingDecoder<T extends Enum<T>>
}
}
private ChannelBuffer cumulation(ChannelHandlerContext ctx) {
ChannelBuffer buf = cumulation.get();
if (buf == null) {
@ -496,13 +511,4 @@ public abstract class ReplayingDecoder<T extends Enum<T>>
}
return buf;
}
private ChannelBuffer cumulation() {
ChannelBuffer cumulation = this.cumulation.get();
if (cumulation == null) {
throw new IllegalStateException(
"checkpoint() should be called in decode() only");
}
return cumulation;
}
}