class GridDirectParser implements GridNioParser {
    /** Message metadata key. */
    static final int MSG_META_KEY = GridNioSessionMetaKey.nextUniqueKey();

    /** Reader metadata key. */
    static final int READER_META_KEY = GridNioSessionMetaKey.nextUniqueKey();

    /** */
    private final IgniteLogger log;

    /** */
    private final MessageFactory msgFactory;

    /** */
    private final GridNioMessageReaderFactory readerFactory;

    /**
     * @param log Logger.
     * @param msgFactory Message factory.
     * @param readerFactory Message reader factory.
     */
    public GridDirectParser(IgniteLogger log, MessageFactory msgFactory, GridNioMessageReaderFactory readerFactory) {
        assert msgFactory != null;
        assert readerFactory != null;

        this.log = log;
        this.msgFactory = msgFactory;
        this.readerFactory = readerFactory;
    }

    /** {@inheritDoc} */
    @Nullable @Override public Object decode(GridNioSession ses, ByteBuffer buf)
        throws IOException, IgniteCheckedException {<FILL_FUNCTION_BODY>}

    /** {@inheritDoc} */
    @Override public ByteBuffer encode(GridNioSession ses, Object msg) throws IOException, IgniteCheckedException {
        // No encoding needed for direct messages.
        throw new UnsupportedEncodingException();
    }
}

        MessageReader reader = ses.meta(READER_META_KEY);

        if (reader == null)
            ses.addMeta(READER_META_KEY, reader = readerFactory.reader(ses, msgFactory));

        Message msg = ses.removeMeta(MSG_META_KEY);

        try {
            if (msg == null && buf.remaining() >= Message.DIRECT_TYPE_SIZE) {
                byte b0 = buf.get();
                byte b1 = buf.get();

                msg = msgFactory.create(makeMessageType(b0, b1));
            }

            boolean finished = false;

            if (msg != null && buf.hasRemaining()) {
                if (reader != null)
                    reader.setCurrentReadClass(msg.getClass());

                finished = msg.readFrom(buf, reader);
            }

            if (finished) {
                if (reader != null)
                    reader.reset();

                return msg;
            }
            else {
                ses.addMeta(MSG_META_KEY, msg);

                return null;
            }
        }
        catch (Throwable e) {
            U.error(log, "Failed to read message [msg=" + msg +
                    ", buf=" + buf +
                    ", reader=" + reader +
                    ", ses=" + ses + "]",
                e);

            throw e;
        }
    
<no_super_class>