//
//  ========================================================================
//  Copyright (c) 1995-2013 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 javax.servlet.ServletInputStream;

import org.eclipse.jetty.io.EofException;
import org.eclipse.jetty.io.RuntimeIOException;
import org.eclipse.jetty.util.ArrayQueue;
import org.eclipse.jetty.util.log.Log;
import org.eclipse.jetty.util.log.Logger;

/**
 * <p>{@link HttpInput} provides an implementation of {@link ServletInputStream} for {@link HttpChannel}.</p>
 * <p>{@link HttpInput} holds a queue of items passed to it by calls to {@link #content(T)}.</p>
 * <p>{@link HttpInput} stores the items directly; if the items contain byte buffers, it does not copy them
 * but simply holds references to the item, thus the caller must organize for those buffers to valid while
 * held by this class.</p>
 * <p>To assist the caller, subclasses may override methods {@link #onContentQueued(T)},
 * {@link #onContentConsumed(T)} and {@link #onAllContentConsumed()} that can be implemented so that the
 * caller will know when buffers are queued and consumed.</p>
 */
public abstract class HttpInput<T> extends ServletInputStream
{
    private final static Logger LOG = Log.getLogger(HttpInput.class);
    private final ArrayQueue<T> _inputQ = new ArrayQueue<>();
    private boolean _earlyEOF;
    private boolean _inputEOF;

    public Object lock()
    {
        return _inputQ.lock();
    }

    public void recycle()
    {
        synchronized (lock())
        {
            T item = _inputQ.peekUnsafe();
            while (item != null)
            {
                _inputQ.pollUnsafe();
                onContentConsumed(item);

                item = _inputQ.peekUnsafe();
                if (item == null)
                    onAllContentConsumed();
            }
            _inputEOF = false;
            _earlyEOF = false;
        }
    }

    @Override
    public int read() throws IOException
    {
        byte[] bytes = new byte[1];
        int read = read(bytes, 0, 1);
        return read < 0 ? -1 : 0xff & bytes[0];
    }

    @Override
    public int available()
    {
        synchronized (lock())
        {
            T item = _inputQ.peekUnsafe();
            if (item == null)
                return 0;
            return remaining(item);
        }
    }

    @Override
    public int read(byte[] b, int off, int len) throws IOException
    {
        T item = null;
        synchronized (lock())
        {
            // Get the current head of the input Q
            item = _inputQ.peekUnsafe();
            
            // Skip empty items at the head of the queue
            while (item != null && remaining(item) == 0)
            {
                _inputQ.pollUnsafe();
                onContentConsumed(item);
                LOG.debug("{} consumed {}", this, item);
                item = _inputQ.peekUnsafe();
                
                // If that was the last item then notify
                if (item==null)
                    onAllContentConsumed();
            }

            // If we have no item
            if (item == null)
            {
                // Was it unexpectedly EOF'd?
                if (isEarlyEOF())
                    throw new EofException();

                // check for EOF
                if (isShutdown())
                {
                    onEOF();
                    return -1;
                }

                // OK then block for some more content
                blockForContent();
                
                // If still not content, we must be closed
                item = _inputQ.peekUnsafe();
                if (item==null)
                {
                    if (isEarlyEOF())
                        throw new EofException();
                    
                    // blockForContent will only return with no 
                    // content if it is closed.
                    if (!isShutdown())
                        LOG.warn("Unexpected !EOF: "+this);

                    onEOF();
                    return -1;
                }
            }
        }
        return get(item, b, off, len);
    }
    protected abstract int remaining(T item);

    protected abstract int get(T item, byte[] buffer, int offset, int length);

    protected abstract void onContentConsumed(T item);

    protected void blockForContent() throws IOException
    {
        synchronized (lock())
        {
            while (_inputQ.isEmpty() && !isShutdown() && !isEarlyEOF())
            {
                try
                {
                    LOG.debug("{} waiting for content", this);
                    lock().wait();
                }
                catch (InterruptedException e)
                {
                    throw (IOException)new InterruptedIOException().initCause(e);
                }
            }
        }
    }

    /* ------------------------------------------------------------ */
    /** Called by this HttpInput to signal new content has been queued
     * @param item
     */
    protected void onContentQueued(T item)
    {
        lock().notify();
    }

    /* ------------------------------------------------------------ */
    /** Called by this HttpInput to signal all available content has been consumed
     */
    protected void onAllContentConsumed()
    {
    }

    /* ------------------------------------------------------------ */
    /** Called by this HttpInput to signal it has reached EOF
     */
    protected void onEOF()
    {
    }

    /* ------------------------------------------------------------ */
    /** Add some content to the input stream
     * @param item
     */
    public void content(T item)
    {
        synchronized (lock())
        {
            // The buffer is not copied here.  This relies on the caller not recycling the buffer
            // until the it is consumed.  The onAllContentConsumed() callback is the signal to the
            // caller that the buffers can be recycled.
            _inputQ.add(item);
            onContentQueued(item);
            LOG.debug("{} queued {}", this, item);
        }
    }

    /* ------------------------------------------------------------ */
    /** This method should be called to signal to the HttpInput
     * that an EOF has arrived before all the expected content.
     * Typically this will result in an EOFException being thrown
     * from a subsequent read rather than a -1 return.
     */
    public void earlyEOF()
    {
        synchronized (lock())
        {
            _earlyEOF = true;
            _inputEOF = true;
            lock().notify();
            LOG.debug("{} early EOF", this);
        }
    }

    /* ------------------------------------------------------------ */
    public boolean isEarlyEOF()
    {
        synchronized (lock())
        {
            return _earlyEOF;
        }
    }

    /* ------------------------------------------------------------ */
    public void shutdown()
    {
        synchronized (lock())
        {
            _inputEOF = true;
            lock().notify();
            LOG.debug("{} shutdown", this);
        }
    }

    /* ------------------------------------------------------------ */
    public boolean isShutdown()
    {
        synchronized (lock())
        {
            return _inputEOF;
        }
    }

    /* ------------------------------------------------------------ */
    public void consumeAll()
    {
        synchronized (lock())
        {
            T item = _inputQ.peekUnsafe();
            while (!isShutdown() && !isEarlyEOF())
            {
                while (item != null)
                {
                    _inputQ.pollUnsafe();
                    onContentConsumed(item);

                    item = _inputQ.peekUnsafe();
                    if (item == null)
                        onAllContentConsumed();
                }

                try
                {
                    blockForContent();
                    item = _inputQ.peekUnsafe();
                    if (item==null)
                        break;
                }
                catch (IOException e)
                {
                    throw new RuntimeIOException(e);
                }
            }
        }
    }
}
