jetty-server/src/main/java/org/eclipse/jetty/server/HttpInput.java - jetty/org.eclipse.jetty.project - Git at Google

 //
 //  ========================================================================
 //  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";
         }
     };

 }
	//
	// ========================================================================
	// 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";
	}
	};

	}