import org.jboss.netty.handler.codec.frame.CorruptedFrameException; //导入依赖的package包/类
/**
 * Called by netty when an exception happens in one of the netty threads
 * (mostly due to what we do in the netty threads)
 */
@Override
public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws Exception {
    Throwable t = e.getCause();
    if (t instanceof CorruptedFrameException || t instanceof TooLongFrameException) {
        LOG.error("Corrupted fram recieved from bookie: " + e.getChannel().getRemoteAddress());
        return;
    }
    if (t instanceof IOException) {
        // these are thrown when a bookie fails, logging them just pollutes
        // the logs (the failure is logged from the listeners on the write
        // operation), so I'll just ignore it here.
        return;
    }

    LOG.fatal("Unexpected exception caught by bookie client channel handler", t);
    // Since we are a library, cant terminate App here, can we?
}
 

import org.jboss.netty.handler.codec.frame.CorruptedFrameException; //导入依赖的package包/类
@Override
protected Object decode(ChannelHandlerContext ctx, Channel channel, ChannelBuffer buffer, State state) throws Exception {
    switch (state) {
    case READ_PREFIX: {
        message = new RpcResponse();
        message.setCode(buffer.readByte());
        int start = buffer.readerIndex();
        int stop = start + actualReadableBytes();
        for (int i = start; i < stop; i++) {
            if (buffer.getByte(i) != RpcConstants.FRAME_START) {
                buffer.skipBytes(i - start);
                break;
            }
        }
        checkpoint(State.READ_CONTENT);
    }
    case READ_CONTENT: {
        boolean eot = false;
        while (!eot) {
            RpcResponse.Line line = new RpcResponse.Line();
            eot = readLine(buffer, line, RpcConstants.MAX_FRAME_LEN);
            message.appendLine(line);
            checkpoint();
        }
        checkpoint(State.READ_PREFIX);
        break;
    }
    default:
        // Should not get here, all cases are handled
        throw new CorruptedFrameException();
    }

    return message;
}
 

import org.jboss.netty.handler.codec.frame.CorruptedFrameException; //导入依赖的package包/类
public static Parameter decodeParameter(ChannelBuffer in) throws CorruptedFrameException {
    Parameter param = null;
    // Fist byte indicates the parameter type
    byte b = in.readByte();
    switch (b) {
    case RpcConstants.FRAME_STOP:
        return null;
    case RpcConstants.PARAM_TYPE_LITERAL: {
        param = new LiteralParameter(decodeField(in));
        break;
    }
    case RpcConstants.PARAM_TYPE_REF: {
        param = new ReferenceParameter(decodeField(in));
        break;
    }
    case RpcConstants.PARAM_TYPE_MAP: {
        param = new MapParameter(decodeMap(in));
        break;
    }
    case RpcConstants.PARAM_TYPE_GLOBAL: {
        param = new GlobalParameter(decodeField(in), decodeField(in));
        break;
    }
    case RpcConstants.PARAM_TYPE_EMPTY: {
        param = new EmptyParameter();
        break;
    }
    case RpcConstants.PARAM_TYPE_STREAM: {
        // TODO: implement me
        break;
    }
    default:
        throw new CorruptedFrameException("Unkown RPC parameter type: '" + String.format("%02x ", b) + "'");
    }
    
    b = in.readByte();
    if (b != RpcConstants.PARAM_STOP) {
        throw new CorruptedFrameException("Expected end of parameter, got '" + String.format("%02x ", b) + "' instead");
    }
    return param;
}
 

import org.jboss.netty.handler.codec.frame.CorruptedFrameException; //导入依赖的package包/类
protected Object decode(ChannelHandlerContext ctx, ChannelBuffer buffer) throws Exception {
    if (lengthBytesToRead > 0) {
        if (lengthBytesToRead > buffer.readableBytes()) {
            lengthBytesToRead -= buffer.readableBytes();
            lengthBuffer.writeBytes(buffer);
            return null;
        } else {
            lengthBuffer.writeBytes(buffer, lengthBytesToRead);
            lengthBytesToRead = 0;
            frameBytesToRead = lengthBuffer.getUnsignedInt(0);

            if (frameBytesToRead < 0) {
                skipFrame = true;
                frameBytesToRead = 0;
                Channels.fireExceptionCaught(ctx, new CorruptedFrameException("negative frame length: " +
                    frameBytesToRead));
            } else if (frameBytesToRead > maxFrameLength) {
                skipFrame = true;
                Channels.fireExceptionCaught(ctx, new TooLongFrameException("frame length exceeds " +
                    maxFrameLength + ": " + frameBytesToRead));
            } else {
                skipFrame = false;
                frameBuffer = getBuffer(ctx.getChannel().getConfig().getBufferFactory(),
                        (int) frameBytesToRead);
            }
        }
    }

    if (frameBytesToRead > buffer.readableBytes()) {
        frameBytesToRead -= buffer.readableBytes();
        if (skipFrame) {
            buffer.skipBytes(buffer.readableBytes());
        } else {
            frameBuffer.writeBytes(buffer);
        }
        return null;
    } else {
        lengthBuffer.setIndex(0, 0);
        lengthBytesToRead = lengthFieldLength;
        if (skipFrame) {
            buffer.skipBytes((int) frameBytesToRead);
            frameBytesToRead = 0;
            return null;
        } else {
            frameBuffer.writeBytes(buffer, (int) frameBytesToRead);
            frameBytesToRead = 0;
            PNPFrame m = PNPFrame.constructMessage(frameBuffer, 0);
            frameBuffer = null;
            return m;
        }
    }
}
 

import org.jboss.netty.handler.codec.frame.CorruptedFrameException; //导入依赖的package包/类
protected Object decode(ChannelHandlerContext ctx, ChannelBuffer buffer) throws Exception {
    if (!firstFrame) {
        this.pnpServerHandler.bytesAvailable();
    }

    if (lengthBytesToRead > 0) {
        if (lengthBytesToRead > buffer.readableBytes()) {
            lengthBytesToRead -= buffer.readableBytes();
            lengthBuffer.writeBytes(buffer);
            return null;
        } else {
            lengthBuffer.writeBytes(buffer, lengthBytesToRead);
            lengthBytesToRead = 0;

            frameBytesToRead = lengthBuffer.getUnsignedInt(0);
            if (frameBytesToRead < 0) {
                skipFrame = true;
                frameBytesToRead = 0;
                Channels.fireExceptionCaught(ctx, new CorruptedFrameException("negative frame length: " +
                    frameBytesToRead));
            } else if (frameBytesToRead > maxFrameLength) {
                skipFrame = true;
                Channels.fireExceptionCaught(ctx, new TooLongFrameException("frame length exceeds " +
                    maxFrameLength + ": " + frameBytesToRead));
            } else {
                skipFrame = false;
                frameBuffer = getBuffer(ctx.getChannel().getConfig().getBufferFactory(),
                        (int) frameBytesToRead);
            }
        }
    }

    if (frameBytesToRead > buffer.readableBytes()) {
        frameBytesToRead -= buffer.readableBytes();
        if (skipFrame) {
            buffer.skipBytes(buffer.readableBytes());
        } else {
            frameBuffer.writeBytes(buffer);
        }
        return null;
    } else {
        lengthBuffer.setIndex(0, 0);
        lengthBytesToRead = lengthFieldLength;
        if (skipFrame) {
            buffer.skipBytes((int) frameBytesToRead);
            frameBytesToRead = 0;
            return null;
        } else {
            frameBuffer.writeBytes(buffer, (int) frameBytesToRead);
            frameBytesToRead = 0;

            PNPFrame m = null;
            try {
                m = PNPFrame.constructMessage(frameBuffer, 0);
            } catch (Exception e) {
                this.pnpServerHandler.clientLeave();
                throw e;
            }

            frameBuffer = null;

            if (firstFrame) {
                if (m.getType() == PNPFrame.MessageType.HEARTBEAT_ADV) {
                    PNPFrameHeartbeatAdvertisement ahFrame = (PNPFrameHeartbeatAdvertisement) m;

                    long heartbeatPeriod = ahFrame.getHeartbeatPeriod();
                    Heartbeater heartbeater = new Heartbeater(ctx.getChannel(), timer, heartbeatPeriod);
                    this.pnpServerHandler.setHeartBeater(heartbeater);

                    this.firstFrame = false;
                    return null; // Do not sent the first frame to the handler
                } else {
                    throw new PNPException("Invalid first frame type must be " +
                        MessageType.HEARTBEAT_ADV + " but is " + m.getType());
                }
            }

            return m;
        }
    }
}