* Fixed compilation errors
* Made sure cumulative buffers are initialized as early as possible
This commit is contained in:
parent
5eec9ac58c
commit
f5fb85a0af
@ -355,7 +355,7 @@ public class DefaultChannelPipeline implements ChannelPipeline {
|
||||
h.beforeAdd(ctx);
|
||||
} catch (Throwable t) {
|
||||
throw new ChannelHandlerLifeCycleException(
|
||||
h.getName() +
|
||||
h.getClass().getName() +
|
||||
".beforeAdd() has thrown an exception; not adding.", t);
|
||||
}
|
||||
}
|
||||
@ -385,11 +385,11 @@ public class DefaultChannelPipeline implements ChannelPipeline {
|
||||
|
||||
if (removed) {
|
||||
throw new ChannelHandlerLifeCycleException(
|
||||
h.getName() +
|
||||
h.getClass().getName() +
|
||||
".afterAdd() has thrown an exception; removed.", t);
|
||||
} else {
|
||||
throw new ChannelHandlerLifeCycleException(
|
||||
h.getName() +
|
||||
h.getClass().getName() +
|
||||
".afterAdd() has thrown an exception; also failed to remove.", t);
|
||||
}
|
||||
}
|
||||
@ -411,7 +411,7 @@ public class DefaultChannelPipeline implements ChannelPipeline {
|
||||
h.beforeRemove(ctx);
|
||||
} catch (Throwable t) {
|
||||
throw new ChannelHandlerLifeCycleException(
|
||||
h.getName() +
|
||||
h.getClass().getName() +
|
||||
".beforeRemove() has thrown an exception; not removing.", t);
|
||||
}
|
||||
}
|
||||
@ -432,7 +432,7 @@ public class DefaultChannelPipeline implements ChannelPipeline {
|
||||
h.afterRemove(ctx);
|
||||
} catch (Throwable t) {
|
||||
throw new ChannelHandlerLifeCycleException(
|
||||
h.getName() +
|
||||
h.getClass().getName() +
|
||||
".afterRemove() has thrown an exception.", t);
|
||||
}
|
||||
}
|
||||
|
@ -36,6 +36,7 @@ import org.jboss.netty.channel.ChannelPipelineCoverage;
|
||||
import org.jboss.netty.channel.ChannelStateEvent;
|
||||
import org.jboss.netty.channel.ChannelUpstreamHandler;
|
||||
import org.jboss.netty.channel.ExceptionEvent;
|
||||
import org.jboss.netty.channel.LifeCycleAwareChannelHandler;
|
||||
import org.jboss.netty.channel.MessageEvent;
|
||||
import org.jboss.netty.channel.SimpleChannelHandler;
|
||||
|
||||
@ -147,11 +148,36 @@ import org.jboss.netty.channel.SimpleChannelHandler;
|
||||
* @apiviz.landmark
|
||||
*/
|
||||
@ChannelPipelineCoverage("one")
|
||||
public abstract class FrameDecoder extends SimpleChannelHandler {
|
||||
public abstract class FrameDecoder
|
||||
extends SimpleChannelHandler implements LifeCycleAwareChannelHandler {
|
||||
|
||||
private final AtomicReference<ChannelBuffer> cumulation =
|
||||
new AtomicReference<ChannelBuffer>();
|
||||
|
||||
|
||||
@Override
|
||||
public void handleUpstream(ChannelHandlerContext ctx, ChannelEvent e)
|
||||
throws Exception {
|
||||
cumulation(ctx);
|
||||
super.handleUpstream(ctx, e);
|
||||
}
|
||||
|
||||
public void beforeAdd(ChannelHandlerContext ctx) throws Exception {
|
||||
cumulation(ctx);
|
||||
}
|
||||
|
||||
public void afterAdd(ChannelHandlerContext ctx) throws Exception {
|
||||
// Unused
|
||||
}
|
||||
|
||||
public void afterRemove(ChannelHandlerContext ctx) throws Exception {
|
||||
// Unused
|
||||
}
|
||||
|
||||
public void beforeRemove(ChannelHandlerContext ctx) throws Exception {
|
||||
// Unused
|
||||
}
|
||||
|
||||
@Override
|
||||
public void messageReceived(
|
||||
ChannelHandlerContext ctx, MessageEvent e) throws Exception {
|
||||
@ -167,7 +193,7 @@ public abstract class FrameDecoder extends SimpleChannelHandler {
|
||||
return;
|
||||
}
|
||||
|
||||
ChannelBuffer cumulation = cumulation(e);
|
||||
ChannelBuffer cumulation = cumulation(ctx);
|
||||
if (cumulation.readable()) {
|
||||
cumulation.discardReadBytes();
|
||||
cumulation.writeBytes(input);
|
||||
@ -256,7 +282,7 @@ public abstract class FrameDecoder extends SimpleChannelHandler {
|
||||
|
||||
private void cleanup(ChannelHandlerContext ctx, ChannelStateEvent e)
|
||||
throws Exception {
|
||||
ChannelBuffer cumulation = cumulation(e);
|
||||
ChannelBuffer cumulation = cumulation(ctx);
|
||||
try {
|
||||
if (cumulation.readable()) {
|
||||
// Make sure all frames are read before notifying a closed channel.
|
||||
@ -274,11 +300,11 @@ public abstract class FrameDecoder extends SimpleChannelHandler {
|
||||
}
|
||||
}
|
||||
|
||||
private ChannelBuffer cumulation(ChannelEvent e) {
|
||||
private ChannelBuffer cumulation(ChannelHandlerContext ctx) {
|
||||
ChannelBuffer buf = cumulation.get();
|
||||
if (buf == null) {
|
||||
buf = ChannelBuffers.dynamicBuffer(
|
||||
e.getChannel().getConfig().getBufferFactory());
|
||||
ctx.getChannel().getConfig().getBufferFactory());
|
||||
if (!cumulation.compareAndSet(null, buf)) {
|
||||
buf = cumulation.get();
|
||||
}
|
||||
|
@ -30,11 +30,13 @@ import java.util.concurrent.atomic.AtomicReference;
|
||||
import org.jboss.netty.buffer.ChannelBuffer;
|
||||
import org.jboss.netty.buffer.ChannelBufferFactory;
|
||||
import org.jboss.netty.channel.Channel;
|
||||
import org.jboss.netty.channel.ChannelEvent;
|
||||
import org.jboss.netty.channel.ChannelHandlerContext;
|
||||
import org.jboss.netty.channel.ChannelPipelineCoverage;
|
||||
import org.jboss.netty.channel.ChannelStateEvent;
|
||||
import org.jboss.netty.channel.Channels;
|
||||
import org.jboss.netty.channel.ExceptionEvent;
|
||||
import org.jboss.netty.channel.LifeCycleAwareChannelHandler;
|
||||
import org.jboss.netty.channel.MessageEvent;
|
||||
import org.jboss.netty.channel.SimpleChannelHandler;
|
||||
import org.jboss.netty.handler.codec.frame.FrameDecoder;
|
||||
@ -212,7 +214,9 @@ import org.jboss.netty.handler.codec.frame.FrameDecoder;
|
||||
* @apiviz.landmark
|
||||
*/
|
||||
@ChannelPipelineCoverage("one")
|
||||
public abstract class ReplayingDecoder<T extends Enum<T>> extends SimpleChannelHandler {
|
||||
public abstract class ReplayingDecoder<T extends Enum<T>>
|
||||
extends SimpleChannelHandler implements LifeCycleAwareChannelHandler {
|
||||
|
||||
|
||||
private final AtomicReference<ChannelBuffer> cumulation =
|
||||
new AtomicReference<ChannelBuffer>();
|
||||
@ -297,6 +301,29 @@ public abstract class ReplayingDecoder<T extends Enum<T>> extends SimpleChannelH
|
||||
return decode(ctx, channel, buffer, state);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void handleUpstream(ChannelHandlerContext ctx, ChannelEvent e)
|
||||
throws Exception {
|
||||
cumulation(ctx);
|
||||
super.handleUpstream(ctx, e);
|
||||
}
|
||||
|
||||
public void beforeAdd(ChannelHandlerContext ctx) throws Exception {
|
||||
cumulation(ctx);
|
||||
}
|
||||
|
||||
public void afterAdd(ChannelHandlerContext ctx) throws Exception {
|
||||
// Unused
|
||||
}
|
||||
|
||||
public void afterRemove(ChannelHandlerContext ctx) throws Exception {
|
||||
// Unused
|
||||
}
|
||||
|
||||
public void beforeRemove(ChannelHandlerContext ctx) throws Exception {
|
||||
// Unused
|
||||
}
|
||||
|
||||
@Override
|
||||
public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
|
||||
throws Exception {
|
||||
|
Loading…
x
Reference in New Issue
Block a user