//
   //  ========================================================================
   //  Copyright (c) 1995-2016 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.Queue;
  import java.util.concurrent.TimeoutException;
  
  import javax.servlet.ReadListener;
  import javax.servlet.ServletInputStream;
  
  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 _contentConsumed;
      private long _blockingTimeoutAt = -1;
  
      public HttpInput(HttpChannelState state)
      {
          _channelState=state;
          if (_channelState.getHttpChannel().getHttpConfiguration().getBlockingTimeout()>0)
              _blockingTimeoutAt=0;
      }
  
      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;
              _contentConsumed = 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()
     {
         _channelState.getHttpChannel().getConnector().getExecutor().execute(_channelState.getHttpChannel());
     }
 
 
     @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 (_blockingTimeoutAt>=0 && !isAsync())
                 _blockingTimeoutAt=System.currentTimeMillis()+getHttpChannelState().getHttpChannel().getHttpConfiguration().getBlockingTimeout();
 
             while(true)
             {
                 Content item = nextContent();
                 if (item!=null)
                 {
                     if (LOG.isDebugEnabled())
                         LOG.debug("{} read {} from {}",this,len,item);
                     int l = get(item, b, off, len);
 
                     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 (_blockingTimeoutAt>=0)
             {
                 timeout=_blockingTimeoutAt-System.currentTimeMillis();
                 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();
 
             if (_blockingTimeoutAt>0 && System.currentTimeMillis()>=_blockingTimeoutAt)
                 throw new TimeoutException();
         }
         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)
         {
             _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()
     {
         return String.format("%s@%x[c=%d,s=%s]",
                 getClass().getSimpleName(),
                 hashCode(),
                 _contentConsumed,
                 _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";
         }
     };
 
 }