@Override
    public void messageSent(NextFilter nextFilter, IoSession session,
            Object message) throws Exception {
        InputStream inputStream = (InputStream) session
                .getAttribute(CURRENT_STREAM);

        if (inputStream == null) {
            nextFilter.messageSent(session, message);
        } else {
            ByteBuffer byteBuffer = getNextByteBuffer(inputStream);

            if (byteBuffer == null) {
                // End of stream reached.
                session.removeAttribute(CURRENT_STREAM);
                WriteFuture writeFuture = (WriteFuture) session
                        .removeAttribute(INITIAL_WRITE_FUTURE);

                // Write queued WriteRequests.
                Queue<? extends WriteRequest> queue = (Queue<? extends WriteRequest>) session
                        .removeAttribute(WRITE_REQUEST_QUEUE);
                if (queue != null) {
                    WriteRequest wr = queue.poll();
                    while (wr != null) {
                        filterWrite(nextFilter, session, wr);
                        wr = queue.poll();
                    }
                }

                writeFuture.setWritten(true);
                nextFilter.messageSent(session, inputStream);
            } else {
                nextFilter.filterWrite(session, new WriteRequest(byteBuffer));
            }
        }
    }