@Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf buf, List<Object> out) {
        if (buf.isReadable()) {
            switch (state()) {
                case READ_LENGTH:
                    this.length = buf.readInt();
                    checkpoint(DecoderState.READ_MSG_TYPE);
                case READ_MSG_TYPE:
                    this.msgType = buf.readShort();
                    checkpoint(DecoderState.READ_MSG);
                case READ_MSG:
                    byte[] frame = new byte[this.length];
                    buf.readBytes(frame);
                    checkpoint(DecoderState.READ_LENGTH);

                    try {
                        switch (Types.values()[this.msgType]) {
                            case TEXT_MESSAGE:
                                out.add(TextMessage.parseFrom(frame));
                                break;
                            case DIRECTLY_COMMUNICATION:
                                out.add(DirectlyCommunication.parseFrom(frame));
                                break;
                            case SUCCESS:
                                out.add(Success.parseFrom(frame));
                                break;
                            case HANDSHAKE:
                                out.add(Handshake.parseFrom(frame));
                                break;
                        }
                    } catch (InvalidProtocolBufferException e) {
                        this.logger.error(e.getMessage(), e);
                    }
            }
        }
    }