package io.netty.handler.codec;

import static io.netty.handler.codec.MessageToMessageEncoder.*;
import io.netty.buffer.ChannelBuffer;
import io.netty.channel.ChannelBufferHolder;
import io.netty.channel.ChannelBufferHolders;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInboundHandlerContext;

public abstract class StreamToMessageDecoder<O> extends ChannelInboundHandlerAdapter<Byte> {

    @Override
    public ChannelBufferHolder<Byte> newInboundBuffer(
            ChannelInboundHandlerContext<Byte> ctx) throws Exception {
        return ChannelBufferHolders.byteBuffer();
    }

    @Override
    public void inboundBufferUpdated(ChannelInboundHandlerContext<Byte> ctx) throws Exception {
        callDecode(ctx);
    }

    @Override
    public void channelInactive(ChannelInboundHandlerContext<Byte> ctx) throws Exception {
        ChannelBuffer in = ctx.in().byteBuffer();
        if (in.readable()) {
            callDecode(ctx);
        }

        try {
            if (unfoldAndAdd(ctx, ctx.nextIn(), decodeLast(ctx, in))) {
                in.discardReadBytes();
                ctx.fireInboundBufferUpdated();
            }
        } catch (Throwable t) {
            if (t instanceof CodecException) {
                ctx.fireExceptionCaught(t);
            } else {
                ctx.fireExceptionCaught(new DecoderException(t));
            }
        }

        ctx.fireChannelInactive();
    }

    private void callDecode(ChannelInboundHandlerContext<Byte> ctx) {
        ChannelBuffer in = ctx.in().byteBuffer();

        boolean decoded = false;
        for (;;) {
            try {
                int oldInputLength = in.readableBytes();
                O o = decode(ctx, in);
                if (o == null) {
                    if (oldInputLength == in.readableBytes()) {
                        break;
                    } else {
                        continue;
                    }
                } else {
                    if (oldInputLength == in.readableBytes()) {
                        throw new IllegalStateException(
                                "decode() did not read anything but decoded a message.");
                    }
                }

                if (unfoldAndAdd(ctx, ctx.nextIn(), o)) {
                    decoded = true;
                } else {
                    break;
                }
            } catch (Throwable t) {
                if (t instanceof CodecException) {
                    ctx.fireExceptionCaught(t);
                } else {
                    ctx.fireExceptionCaught(new DecoderException(t));
                }
            }
        }

        if (decoded) {
            in.discardReadBytes();
            ctx.fireInboundBufferUpdated();
        }
    }

    public abstract O decode(ChannelInboundHandlerContext<Byte> ctx, ChannelBuffer in) throws Exception;

    public O decodeLast(ChannelInboundHandlerContext<Byte> ctx, ChannelBuffer in) throws Exception {
        return decode(ctx, in);
    }
}

@@ -219,16 +219,63 @@
 olders;%0A
+import io.netty.channel.ChannelInboundHandler;%0A
 import i
@@ -316,24 +316,24 @@
 lerAdapter;%0A
-
 import io.ne
@@ -373,16 +373,57 @@
 Context;
+%0Aimport io.netty.channel.ChannelPipeline;
 %0A%0Apublic
@@ -510,16 +510,69 @@
 yte%3E %7B%0A%0A
+    private ChannelInboundHandlerContext%3CByte%3E ctx;%0A%0A
     @Ove
@@ -632,16 +632,16 @@
 Buffer(%0A
-
         
@@ -695,32 +695,56 @@
 ows Exception %7B%0A
+        this.ctx = ctx;%0A
         return C
@@ -2943,24 +2943,963 @@
    %7D%0A    %7D%0A%0A
+    /**%0A     * Replace this decoder in the %7B@link ChannelPipeline%7D with the given handler.%0A     * All remaining bytes in the inbound buffer will be forwarded to the new handler's%0A     * inbound buffer.%0A     */%0A    public void replace(String newHandlerName, ChannelInboundHandler%3CByte%3E newHandler) %7B%0A        if (!ctx.channel().eventLoop().inEventLoop()) %7B%0A            throw new IllegalStateException(%22not in event loop%22);%0A        %7D%0A%0A        // We do not use ChannelPipeline.replace() here so that the current context points%0A        // the new handler.%0A        ctx.pipeline().addAfter(ctx.name(), newHandlerName, newHandler);%0A%0A        ChannelBuffer in = ctx.in().byteBuffer();%0A        try %7B%0A            if (in.readable()) %7B%0A                ctx.nextIn().byteBuffer().writeBytes(ctx.in().byteBuffer());%0A                ctx.fireInboundBufferUpdated();%0A            %7D%0A        %7D finally %7B%0A            ctx.pipeline().remove(this);%0A        %7D%0A    %7D%0A%0A
     public a

Add StreamToMessageDecoder.replace() (#332)