jetty-server/src/main/java/org/eclipse/jetty/server/HttpInput.java

//
//  ========================================================================
//  Copyright (c) 1995-2017 Mort Bay Consulting Pty. Ltd.
//  ------------------------------------------------------------------------
//  All rights reserved. This program and the accompanying materials
//  are made available under the terms of the Eclipse Public License v1.0
//  and Apache License v2.0 which accompanies this distribution.
//
//      The Eclipse Public License is available at
//      http://www.eclipse.org/legal/epl-v10.html
//
//      The Apache License v2.0 is available at
//      http://www.opensource.org/licenses/apache2.0.php
//
//  You may elect to redistribute this code under either of these licenses.
//  ========================================================================
//

package org.eclipse.jetty.server;

import java.io.IOException;
import java.io.InterruptedIOException;
import java.nio.ByteBuffer;
import java.util.ArrayDeque;
import java.util.Deque;
import java.util.Objects;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import javax.servlet.ReadListener;
import javax.servlet.ServletInputStream;

import org.eclipse.jetty.http.BadMessageException;
import org.eclipse.jetty.http.HttpStatus;
import org.eclipse.jetty.io.EofException;
import org.eclipse.jetty.io.RuntimeIOException;
import org.eclipse.jetty.util.BufferUtil;
import org.eclipse.jetty.util.Callback;
import org.eclipse.jetty.util.log.Log;
import org.eclipse.jetty.util.log.Logger;

/**
 * {@link HttpInput} provides an implementation of {@link ServletInputStream} for {@link HttpChannel}.
 * <p>
 * Content may arrive in patterns such as [content(), content(), messageComplete()] so that this class
 * maintains two states: the content state that tells whether there is content to consume and the EOF
 * state that tells whether an EOF has arrived.
 * Only once the content has been consumed the content state is moved to the EOF state.
 */
public class HttpInput extends ServletInputStream implements Runnable
{
    private final static Logger LOG = Log.getLogger(HttpInput.class);
    private final static Content EOF_CONTENT = new EofContent("EOF");
    private final static Content EARLY_EOF_CONTENT = new EofContent("EARLY_EOF");

    private final byte[] _oneByteBuffer = new byte[1];
    private final Deque<Content> _inputQ = new ArrayDeque<>();
    private final HttpChannelState _channelState;
    private ReadListener _listener;
    private State _state = STREAM;
    private long _firstByteTimeStamp = -1;
    private long _contentArrived;
    private long _contentConsumed;
    private long _blockUntil;

    public HttpInput(HttpChannelState state)
    {
        _channelState = state;
    }

    protected HttpChannelState getHttpChannelState()
    {
        return _channelState;
    }

    public void recycle()
    {
        synchronized (_inputQ)
        {
            Content item = _inputQ.poll();
            while (item != null)
            {
                item.failed(null);
                item = _inputQ.poll();
            }
            _listener = null;
            _state = STREAM;
            _contentArrived = 0;
            _contentConsumed = 0;
            _firstByteTimeStamp = -1;
            _blockUntil = 0;
        }
    }

    @Override
    public int available()
    {
        int available = 0;
        boolean woken = false;
        synchronized (_inputQ)
        {
            Content content = _inputQ.peek();
            if (content == null)
            {
                try
                {
                    produceContent();
                }
                catch (IOException e)
                {
                    woken = failed(e);
                }
                content = _inputQ.peek();
            }

            if (content != null)
                available = remaining(content);
        }

        if (woken)
            wake();
        return available;
    }

    private void wake()
    {
        HttpChannel channel = _channelState.getHttpChannel();
        Executor executor = channel.getConnector().getServer().getThreadPool();
        executor.execute(channel);
    }

    private long getBlockingTimeout()
    {
        return getHttpChannelState().getHttpChannel().getHttpConfiguration().getBlockingTimeout();
    }

    @Override
    public int read() throws IOException
    {
        int read = read(_oneByteBuffer, 0, 1);
        if (read == 0)
            throw new IllegalStateException("unready read=0");
        return read < 0 ? -1 : _oneByteBuffer[0] & 0xFF;
    }

    @Override
    public int read(byte[] b, int off, int len) throws IOException
    {
        synchronized (_inputQ)
        {
            if (!isAsync())
            {
                if (_blockUntil == 0)
                {
                    long blockingTimeout = getBlockingTimeout();
                    if (blockingTimeout > 0)
                        _blockUntil = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(blockingTimeout);
                }
            }

            long minRequestDataRate = _channelState.getHttpChannel().getHttpConfiguration().getMinRequestDataRate();
            if (minRequestDataRate > 0 && _firstByteTimeStamp != -1)
            {
                long period = System.nanoTime() - _firstByteTimeStamp;
                if (period > 0)
                {
                    long minimum_data = minRequestDataRate * TimeUnit.NANOSECONDS.toMillis(period) / TimeUnit.SECONDS.toMillis(1);
                    if (_contentArrived < minimum_data)
                        throw new BadMessageException(HttpStatus.REQUEST_TIMEOUT_408, String.format("Request data rate < %d B/s", minRequestDataRate));
                }
            }

            while (true)
            {
                Content item = nextContent();
                if (item != null)
                {
                    int l = get(item, b, off, len);
                    if (LOG.isDebugEnabled())
                        LOG.debug("{} read {} from {}", this, l, item);

                    consumeNonContent();

                    return l;
                }

                if (!_state.blockForContent(this))
                    return _state.noContent();
            }
        }
    }

    /**
     * Called when derived implementations should attempt to
     * produce more Content and add it via {@link #addContent(Content)}.
     * For protocols that are constantly producing (eg HTTP2) this can
     * be left as a noop;
     *
     * @throws IOException if unable to produce content
     */
    protected void produceContent() throws IOException
    {
    }

    /**
     * Get the next content from the inputQ, calling {@link #produceContent()}
     * if need be.  EOF is processed and state changed.
     *
     * @return the content or null if none available.
     * @throws IOException if retrieving the content fails
     */
    protected Content nextContent() throws IOException
    {
        Content content = pollContent();
        if (content == null && !isFinished())
        {
            produceContent();
            content = pollContent();
        }
        return content;
    }

    /**
     * Poll the inputQ for Content.
     * Consumed buffers and {@link PoisonPillContent}s are removed and
     * EOF state updated if need be.
     *
     * @return Content or null
     */
    protected Content pollContent()
    {
        // Items are removed only when they are fully consumed.
        Content content = _inputQ.peek();
        // Skip consumed items at the head of the queue.
        while (content != null && remaining(content) == 0)
        {
            _inputQ.poll();
            content.succeeded();
            if (LOG.isDebugEnabled())
                LOG.debug("{} consumed {}", this, content);

            if (content == EOF_CONTENT)
            {
                if (_listener == null)
                    _state = EOF;
                else
                {
                    _state = AEOF;
                    boolean woken = _channelState.onReadReady(); // force callback?
                    if (woken)
                        wake();
                }
            }
            else if (content == EARLY_EOF_CONTENT)
                _state = EARLY_EOF;

            content = _inputQ.peek();
        }

        return content;
    }

    /**
     */
    protected void consumeNonContent()
    {
        // Items are removed only when they are fully consumed.
        Content content = _inputQ.peek();
        // Skip consumed items at the head of the queue.
        while (content != null && remaining(content) == 0)
        {
            // Defer EOF until read
            if (content instanceof EofContent)
                break;

            // Consume all other empty content
            _inputQ.poll();
            content.succeeded();
            if (LOG.isDebugEnabled())
                LOG.debug("{} consumed {}", this, content);
            content = _inputQ.peek();
        }
    }

    /**
     * Get the next readable from the inputQ, calling {@link #produceContent()}
     * if need be. EOF is NOT processed and state is not changed.
     *
     * @return the content or EOF or null if none available.
     * @throws IOException if retrieving the content fails
     */
    protected Content nextReadable() throws IOException
    {
        Content content = pollReadable();
        if (content == null && !isFinished())
        {
            produceContent();
            content = pollReadable();
        }
        return content;
    }

    /**
     * Poll the inputQ for Content or EOF.
     * Consumed buffers and non EOF {@link PoisonPillContent}s are removed.
     * EOF state is not updated.
     *
     * @return Content, EOF or null
     */
    protected Content pollReadable()
    {
        // Items are removed only when they are fully consumed.
        Content content = _inputQ.peek();

        // Skip consumed items at the head of the queue except EOF
        while (content != null)
        {
            if (content == EOF_CONTENT || content == EARLY_EOF_CONTENT || remaining(content) > 0)
                return content;

            _inputQ.poll();
            content.succeeded();
            if (LOG.isDebugEnabled())
                LOG.debug("{} consumed {}", this, content);
            content = _inputQ.peek();
        }

        return null;
    }

    /**
     * @param item the content
     * @return how many bytes remain in the given content
     */
    protected int remaining(Content item)
    {
        return item.remaining();
    }

    /**
     * Copies the given content into the given byte buffer.
     *
     * @param content the content to copy from
     * @param buffer  the buffer to copy into
     * @param offset  the buffer offset to start copying from
     * @param length  the space available in the buffer
     * @return the number of bytes actually copied
     */
    protected int get(Content content, byte[] buffer, int offset, int length)
    {
        int l = Math.min(content.remaining(), length);
        content.getContent().get(buffer, offset, l);
        _contentConsumed += l;
        return l;
    }

    /**
     * Consumes the given content.
     * Calls the content succeeded if all content consumed.
     *
     * @param content the content to consume
     * @param length  the number of bytes to consume
     */
    protected void skip(Content content, int length)
    {
        int l = Math.min(content.remaining(), length);
        ByteBuffer buffer = content.getContent();
        buffer.position(buffer.position() + l);
        _contentConsumed += l;
        if (l > 0 && !content.hasContent())
            pollContent(); // hungry succeed

    }

    /**
     * Blocks until some content or some end-of-file event arrives.
     *
     * @throws IOException if the wait is interrupted
     */
    protected void blockForContent() throws IOException
    {
        try
        {
            long timeout = 0;
            if (_blockUntil != 0)
            {
                timeout = TimeUnit.NANOSECONDS.toMillis(_blockUntil - System.nanoTime());
                if (timeout <= 0)
                    throw new TimeoutException();
            }

            if (LOG.isDebugEnabled())
                LOG.debug("{} blocking for content timeout={}", this, timeout);
            if (timeout > 0)
                _inputQ.wait(timeout);
            else
                _inputQ.wait();

            // TODO: cannot return unless there is content or timeout,
            // TODO: so spurious wakeups are not handled correctly.

            if (_blockUntil != 0 && TimeUnit.NANOSECONDS.toMillis(_blockUntil - System.nanoTime()) <= 0)
                throw new TimeoutException(String.format("Blocking timeout %d ms", getBlockingTimeout()));
        }
        catch (Throwable e)
        {
            throw (IOException)new InterruptedIOException().initCause(e);
        }
    }

    /**
     * Adds some content to the start of this input stream.
     * <p>Typically used to push back content that has
     * been read, perhaps mutated.  The bytes prepended are
     * deducted for the contentConsumed total</p>
     *
     * @param item the content to add
     * @return true if content channel woken for read
     */
    public boolean prependContent(Content item)
    {
        boolean woken = false;
        synchronized (_inputQ)
        {
            _inputQ.push(item);
            _contentConsumed -= item.remaining();
            if (LOG.isDebugEnabled())
                LOG.debug("{} prependContent {}", this, item);

            if (_listener == null)
                _inputQ.notify();
            else
                woken = _channelState.onReadPossible();
        }

        return woken;
    }

    /**
     * Adds some content to this input stream.
     *
     * @param item the content to add
     * @return true if content channel woken for read
     */
    public boolean addContent(Content item)
    {
        boolean woken = false;
        synchronized (_inputQ)
        {
            if (_firstByteTimeStamp == -1)
                _firstByteTimeStamp = System.nanoTime();
            _contentArrived += item.remaining();
            _inputQ.offer(item);
            if (LOG.isDebugEnabled())
                LOG.debug("{} addContent {}", this, item);

            if (_listener == null)
                _inputQ.notify();
            else
                woken = _channelState.onReadPossible();
        }

        return woken;
    }

    public boolean hasContent()
    {
        synchronized (_inputQ)
        {
            return _inputQ.size() > 0;
        }
    }

    public void unblock()
    {
        synchronized (_inputQ)
        {
            _inputQ.notify();
        }
    }

    public long getContentConsumed()
    {
        synchronized (_inputQ)
        {
            return _contentConsumed;
        }
    }

    /**
     * This method should be called to signal that an EOF has been
     * detected before all the expected content arrived.
     * <p>
     * Typically this will result in an EOFException being thrown
     * from a subsequent read rather than a -1 return.
     *
     * @return true if content channel woken for read
     */
    public boolean earlyEOF()
    {
        return addContent(EARLY_EOF_CONTENT);
    }

    /**
     * This method should be called to signal that all the expected
     * content arrived.
     *
     * @return true if content channel woken for read
     */
    public boolean eof()
    {
        return addContent(EOF_CONTENT);
    }

    public boolean consumeAll()
    {
        synchronized (_inputQ)
        {
            try
            {
                while (!isFinished())
                {
                    Content item = nextContent();
                    if (item == null)
                        break; // Let's not bother blocking

                    skip(item, remaining(item));
                }
                return isFinished() && !isError();
            }
            catch (IOException e)
            {
                LOG.debug(e);
                return false;
            }
        }
    }

    public boolean isError()
    {
        synchronized (_inputQ)
        {
            return _state instanceof ErrorState;
        }
    }

    public boolean isAsync()
    {
        synchronized (_inputQ)
        {
            return _state == ASYNC;
        }
    }

    @Override
    public boolean isFinished()
    {
        synchronized (_inputQ)
        {
            return _state instanceof EOFState;
        }
    }

    @Override
    public boolean isReady()
    {
        try
        {
            synchronized (_inputQ)
            {
                if (_listener == null)
                    return true;
                if (_state instanceof EOFState)
                    return true;
                if (nextReadable() != null)
                    return true;

                _channelState.onReadUnready();
            }
            return false;
        }
        catch (IOException e)
        {
            LOG.ignore(e);
            return true;
        }
    }

    @Override
    public void setReadListener(ReadListener readListener)
    {
        readListener = Objects.requireNonNull(readListener);
        boolean woken = false;
        try
        {
            synchronized (_inputQ)
            {
                if (_listener != null)
                    throw new IllegalStateException("ReadListener already set");
                if (_state != STREAM)
                    throw new IllegalStateException("State " + STREAM + " != " + _state);

                _state = ASYNC;
                _listener = readListener;
                boolean content = nextContent() != null;

                if (content)
                    woken = _channelState.onReadReady();
                else
                    _channelState.onReadUnready();
            }
        }
        catch (IOException e)
        {
            throw new RuntimeIOException(e);
        }

        if (woken)
            wake();
    }

    public boolean failed(Throwable x)
    {
        boolean woken = false;
        synchronized (_inputQ)
        {
            if (_state instanceof ErrorState)
                LOG.warn(x);
            else
                _state = new ErrorState(x);

            if (_listener == null)
                _inputQ.notify();
            else
                woken = _channelState.onReadPossible();
        }

        return woken;
    }

    /*
     * <p>
     * While this class is-a Runnable, it should never be dispatched in it's own thread. It is a
     * runnable only so that the calling thread can use {@link ContextHandler#handle(Runnable)}
     * to setup classloaders etc.
     * </p>
     */
    @Override
    public void run()
    {
        final Throwable error;
        final ReadListener listener;
        boolean aeof = false;

        synchronized (_inputQ)
        {
            if (_state == EOF)
                return;

            if (_state == AEOF)
            {
                _state = EOF;
                aeof = true;
            }

            listener = _listener;
            error = _state instanceof ErrorState ? ((ErrorState)_state).getError() : null;
        }

        try
        {
            if (error != null)
            {
                _channelState.getHttpChannel().getResponse().getHttpFields().add(HttpConnection.CONNECTION_CLOSE);
                listener.onError(error);
            }
            else if (aeof)
            {
                listener.onAllDataRead();
            }
            else
            {
                listener.onDataAvailable();
            }
        }
        catch (Throwable e)
        {
            LOG.warn(e.toString());
            LOG.debug(e);
            try
            {
                if (aeof || error == null)
                {
                    _channelState.getHttpChannel().getResponse().getHttpFields().add(HttpConnection.CONNECTION_CLOSE);
                    listener.onError(e);
                }
            }
            catch (Throwable e2)
            {
                LOG.warn(e2.toString());
                LOG.debug(e2);
                throw new RuntimeIOException(e2);
            }
        }
    }

    @Override
    public String toString()
    {
        State state;
        long consumed;
        int q;
        Content content;
        synchronized (_inputQ)
        {
            state = _state;
            consumed = _contentConsumed;
            q = _inputQ.size();
            content = _inputQ.peekFirst();
        }
        return String.format("%s@%x[c=%d,q=%d,[0]=%s,s=%s]",
                getClass().getSimpleName(),
                hashCode(),
                consumed,
                q,
                content,
                state);
    }

    public static class PoisonPillContent extends Content
    {
        private final String _name;

        public PoisonPillContent(String name)
        {
            super(BufferUtil.EMPTY_BUFFER);
            _name = name;
        }

        @Override
        public String toString()
        {
            return _name;
        }
    }

    public static class EofContent extends PoisonPillContent
    {
        EofContent(String name)
        {
            super(name);
        }
    }

    public static class Content implements Callback
    {
        private final ByteBuffer _content;

        public Content(ByteBuffer content)
        {
            _content = content;
        }

        @Override
        public boolean isNonBlocking()
        {
            return true;
        }


        public ByteBuffer getContent()
        {
            return _content;
        }

        public boolean hasContent()
        {
            return _content.hasRemaining();
        }

        public int remaining()
        {
            return _content.remaining();
        }

        @Override
        public String toString()
        {
            return String.format("Content@%x{%s}", hashCode(), BufferUtil.toDetailString(_content));
        }
    }


    protected static abstract class State
    {
        public boolean blockForContent(HttpInput in) throws IOException
        {
            return false;
        }

        public int noContent() throws IOException
        {
            return -1;
        }
    }

    protected static class EOFState extends State
    {
    }

    protected class ErrorState extends EOFState
    {
        final Throwable _error;

        ErrorState(Throwable error)
        {
            _error = error;
        }

        public Throwable getError()
        {
            return _error;
        }

        @Override
        public int noContent() throws IOException
        {
            if (_error instanceof IOException)
                throw (IOException)_error;
            throw new IOException(_error);
        }

        @Override
        public String toString()
        {
            return "ERROR:" + _error;
        }
    }

    protected static final State STREAM = new State()
    {
        @Override
        public boolean blockForContent(HttpInput input) throws IOException
        {
            input.blockForContent();
            return true;
        }

        @Override
        public String toString()
        {
            return "STREAM";
        }
    };

    protected static final State ASYNC = new State()
    {
        @Override
        public int noContent() throws IOException
        {
            return 0;
        }

        @Override
        public String toString()
        {
            return "ASYNC";
        }
    };

    protected static final State EARLY_EOF = new EOFState()
    {
        @Override
        public int noContent() throws IOException
        {
            throw new EofException("Early EOF");
        }

        @Override
        public String toString()
        {
            return "EARLY_EOF";
        }
    };

    protected static final State EOF = new EOFState()
    {
        @Override
        public String toString()
        {
            return "EOF";
        }
    };

    protected static final State AEOF = new EOFState()
    {
        @Override
        public String toString()
        {
            return "AEOF";
        }
    };
}