//
   //  ========================================================================
   //  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.EOFException;
  import java.io.IOException;
  import java.io.InputStream;
  import java.nio.ByteBuffer;
  import java.nio.channels.ReadableByteChannel;
  import javax.servlet.RequestDispatcher;
  import javax.servlet.ServletOutputStream;
  import javax.servlet.ServletRequest;
  import javax.servlet.ServletResponse;
  
  import org.eclipse.jetty.http.HttpContent;
  import org.eclipse.jetty.http.HttpHeader;
  import org.eclipse.jetty.io.EofException;
  import org.eclipse.jetty.util.BlockingCallback;
  import org.eclipse.jetty.util.BufferUtil;
  import org.eclipse.jetty.util.Callback;
  import org.eclipse.jetty.util.IteratingCallback;
  import org.eclipse.jetty.util.log.Log;
  import org.eclipse.jetty.util.log.Logger;
  import org.eclipse.jetty.util.resource.Resource;
  
  /**
   * <p>{@link HttpOutput} implements {@link ServletOutputStream}
   * as required by the Servlet specification.</p>
   * <p>{@link HttpOutput} buffers content written by the application until a
   * further write will overflow the buffer, at which point it triggers a commit
   * of the response.</p>
   * <p>{@link HttpOutput} can be closed and reopened, to allow requests included
   * via {@link RequestDispatcher#include(ServletRequest, ServletResponse)} to
   * close the stream, to be reopened after the inclusion ends.</p>
   */
  public class HttpOutput extends ServletOutputStream
  {
      private static Logger LOG = Log.getLogger(HttpOutput.class);
      private final HttpChannel<?> _channel;
      private boolean _closed;
      private long _written;
      private ByteBuffer _aggregate;
      private int _bufferSize;
  
      public HttpOutput(HttpChannel<?> channel)
      {
          _channel = channel;
          _bufferSize = _channel.getHttpConfiguration().getOutputBufferSize();
      }
  
      public boolean isWritten()
      {
          return _written > 0;
      }
  
      public long getWritten()
      {
          return _written;
      }
  
      public void reset()
      {
          _written = 0;
          reopen();
      }
  
      public void reopen()
      {
          _closed = false;
      }
  
      /** Called by the HttpChannel if the output was closed
       * externally (eg by a 500 exception handling).
       */
      void closed()
      {
          if (!_closed)
          {
              _closed = true;
              try
              {
                  _channel.getResponse().closeOutput(); 
              }
             catch(IOException e)
             {
                 _channel.failed();
                 LOG.ignore(e);
             }
             releaseBuffer();
         }
     }
 
     @Override
     public void close()
     {
         if (!isClosed())
         {
             try
             {
                 if (BufferUtil.hasContent(_aggregate))
                     _channel.write(_aggregate, !_channel.getResponse().isIncluding());
                 else
                     _channel.write(BufferUtil.EMPTY_BUFFER, !_channel.getResponse().isIncluding());
             }
             catch(IOException e)
             {
                 _channel.failed();
                 LOG.ignore(e);
             }
         }
         closed();
     }
 
     private void releaseBuffer()
     {
         if (_aggregate != null)
         {
             _channel.getConnector().getByteBufferPool().release(_aggregate);
             _aggregate = null;
         }
     }
 
     public boolean isClosed()
     {
         return _closed;
     }
 
     @Override
     public void flush() throws IOException
     {
         if (isClosed())
             return;
 
         if (BufferUtil.hasContent(_aggregate))
             _channel.write(_aggregate, false);
         else
             _channel.write(BufferUtil.EMPTY_BUFFER, false);
     }
 
     public boolean isAllContentWritten()
     {
         Response response=_channel.getResponse();
         return response.isAllContentWritten(_written);
     }
     
     public void closeOutput() throws IOException
     {
         _channel.getResponse().closeOutput();
     }
 
     @Override
     public void write(byte[] b, int off, int len) throws IOException
     {  
         if (isClosed())
             throw new EofException("Closed");
 
         _written+=len;
         boolean complete=_channel.getResponse().isAllContentWritten(_written);
         int capacity = getBufferSize();
 
         // Should we aggregate?
         if (!complete && len<=capacity/4)
         {
             if (_aggregate == null)
                 _aggregate = _channel.getByteBufferPool().acquire(capacity, false);
 
             // YES - fill the aggregate with content from the buffer
             int filled = BufferUtil.fill(_aggregate, b, off, len);
 
             // return if we are not complete, not full and filled all the content
             if (!complete && filled == len && !BufferUtil.isFull(_aggregate))
                 return;
 
             // adjust offset/length
             off += filled;
             len -= filled;
         }
 
         // flush any content from the aggregate
         if (BufferUtil.hasContent(_aggregate))
         {
             _channel.write(_aggregate, complete && len==0);
 
             // should we fill aggregate again from the buffer?
             if (len>0 && !complete && len<=_aggregate.capacity()/4)
             {
                 BufferUtil.append(_aggregate, b, off, len);
                 return;
             }
         }
 
         // write any remaining content in the buffer directly
         if (len>0)
             // pass as readonly to avoid space stealing optimisation in HttpConnection 
             _channel.write(ByteBuffer.wrap(b, off, len).asReadOnlyBuffer(), complete);
         else if (complete)
             _channel.write(BufferUtil.EMPTY_BUFFER,complete);
 
         if (complete)
             closed();
     }
 
 
     @Override
     public void write(int b) throws IOException
     {
         if (isClosed())
             throw new EOFException("Closed");
 
         // Allocate an aggregate buffer.
         // Never direct as it is slow to do little writes to a direct buffer.
         if (_aggregate == null)
             _aggregate = _channel.getByteBufferPool().acquire(getBufferSize(), false);
 
         BufferUtil.append(_aggregate, (byte)b);
         _written++;
 
         boolean complete=_channel.getResponse().isAllContentWritten(_written);
         
         // Check if all written or full
         if (complete ||  BufferUtil.isFull(_aggregate))
         {
             BlockingCallback callback = _channel.getWriteBlockingCallback();
             _channel.write(_aggregate, false, callback);
             callback.block();
             if (complete)
                 closed();
         }
     }
 
     @Override
     public void print(String s) throws IOException
     {
         if (isClosed())
             throw new IOException("Closed");
 
         write(s.getBytes(_channel.getResponse().getCharacterEncoding()));
     }
 
     /* ------------------------------------------------------------ */
     /** Set headers and send content.
      * @deprecated Use {@link Response#setHeaders(HttpContent)} and {@link #sendContent(HttpContent)} instead.
      * @param content
      * @throws IOException
      */
     @Deprecated
     public void sendContent(Object content) throws IOException
     {
         final BlockingCallback callback =_channel.getWriteBlockingCallback();
 
         if (content instanceof HttpContent)
         {
             _channel.getResponse().setHeaders((HttpContent)content);
             sendContent((HttpContent)content,callback);
         }
         else if (content instanceof Resource)
         {
             Resource resource = (Resource)content;
             _channel.getResponse().getHttpFields().putDateField(HttpHeader.LAST_MODIFIED, resource.lastModified());
             
             ReadableByteChannel in=((Resource)content).getReadableByteChannel();
             if (in!=null)
                 sendContent(in,callback);
             else
                 sendContent(resource.getInputStream(),callback);
         }
         else if (content instanceof ByteBuffer)
         {
             sendContent((ByteBuffer)content,callback);
         }
         else if (content instanceof ReadableByteChannel)
         {
             sendContent((ReadableByteChannel)content,callback);
         }
         else if (content instanceof InputStream)
         {
             sendContent((InputStream)content,callback);
         }
         else
             callback.failed(new IllegalArgumentException("unknown content type "+content.getClass()));
 
         callback.block();
     }
 
     /* ------------------------------------------------------------ */
     /** Blocking send of content.
      * @param content The content to send.
      * @throws IOException
      */
     public void sendContent(ByteBuffer content) throws IOException
     {
         final BlockingCallback callback =_channel.getWriteBlockingCallback();
         if (content.hasArray()&&content.limit()<content.capacity())
             content=content.asReadOnlyBuffer();
         _channel.write(content,true,callback);
         callback.block();
     }
 
     /* ------------------------------------------------------------ */
     /** Blocking send of content.
      * @param in The content to send
      * @throws IOException
      */
     public void sendContent(InputStream in) throws IOException
     {
         final BlockingCallback callback =_channel.getWriteBlockingCallback();
         new InputStreamWritingCB(in,callback).iterate();
         callback.block();
     }
 
     /* ------------------------------------------------------------ */
     /** Blocking send of content.
      * @param in The content to send
      * @throws IOException
      */
     public void sendContent(ReadableByteChannel in) throws IOException
     {
         final BlockingCallback callback =_channel.getWriteBlockingCallback();
         new ReadableByteChannelWritingCB(in,callback).iterate();
         callback.block();
     }
     
 
     /* ------------------------------------------------------------ */
     /** Blocking send of content.
      * @param content The content to send
      * @throws IOException
      */
     public void sendContent(HttpContent content) throws IOException
     {
         final BlockingCallback callback =_channel.getWriteBlockingCallback();
         sendContent(content,callback);
         callback.block();
     }
    
     /* ------------------------------------------------------------ */
     /** Asynchronous send of content.
      * @param content The content to send
      * @param callback The callback to use to notify success or failure
      */
     public void sendContent(ByteBuffer content, final Callback callback)
     {
         if (content.hasArray()&&content.limit()<content.capacity())
             content=content.asReadOnlyBuffer();
         _channel.write(content,true,new Callback()
         {
             @Override
             public void succeeded()
             {
                 closed();
                 callback.succeeded();
             }
 
             @Override
             public void failed(Throwable x)
             {
                 callback.failed(x);
             }            
         });
     }
 
     /* ------------------------------------------------------------ */
     /** Asynchronous send of content.
      * @param in The content to send as a stream.  The stream will be closed 
      * after reading all content.
      * @param callback The callback to use to notify success or failure
      */
     public void sendContent(InputStream in, Callback callback)
     {
         new InputStreamWritingCB(in,callback).iterate();
     }
 
     /* ------------------------------------------------------------ */
     /** Asynchronous send of content.
      * @param in The content to send as a channel.  The channel will be closed 
      * after reading all content.
      * @param callback The callback to use to notify success or failure
      */
     public void sendContent(ReadableByteChannel in, Callback callback)
     {
         new ReadableByteChannelWritingCB(in,callback).iterate();
     }
 
     /* ------------------------------------------------------------ */
     /** Asynchronous send of content.
      * @param httpContent The content to send
      * @param callback The callback to use to notify success or failure
      */
     public void sendContent(HttpContent httpContent, Callback callback) throws IOException
     {
         if (isClosed())
             throw new IOException("Closed");
         if (BufferUtil.hasContent(_aggregate))
             throw new IOException("written");
         if (_channel.isCommitted())
             throw new IOException("committed");
             
         ByteBuffer buffer= _channel.useDirectBuffers()?httpContent.getDirectBuffer():null;
         if (buffer == null)
             buffer = httpContent.getIndirectBuffer();
         
         if (buffer!=null)
         {
             sendContent(buffer,callback);
             return;
         }
         
         ReadableByteChannel rbc=httpContent.getReadableByteChannel();
         if (rbc!=null)
         {
             // Close of the rbc is done by the async sendContent
             sendContent(rbc,callback);
             return;
         }
            
         InputStream in = httpContent.getInputStream();
         if ( in!=null )
         {
             sendContent(in,callback);
             return;
         }
 
         callback.failed(new IllegalArgumentException("unknown content for "+httpContent));
     }
 
     public int getBufferSize()
     {
         return _bufferSize;
     }
 
     public void setBufferSize(int size)
     {
         this._bufferSize = size;
     }
 
     public void resetBuffer()
     {
         if (BufferUtil.hasContent(_aggregate))
             BufferUtil.clear(_aggregate);
     }
     
     
     /* ------------------------------------------------------------ */
     /** An iterating callback that will take content from an 
      * InputStream and write it to the associated {@link HttpChannel}.
      * A non direct buffer of size {@link HttpOutput#getBufferSize()} is used. 
      * This callback is passed to the {@link HttpChannel#write(ByteBuffer, boolean, Callback)} to
      * be notified as each buffer is written and only once all the input is consumed will the 
      * wrapped {@link Callback#succeeded()} method be called. 
      */
     private class InputStreamWritingCB extends IteratingCallback
     {
         private final InputStream _in;
         private final ByteBuffer _buffer;
         private boolean _eof;
         
         public InputStreamWritingCB(InputStream in, Callback callback)
         {          
             super(callback);
             _in=in;
             _buffer = _channel.getByteBufferPool().acquire(getBufferSize(), false);
         }
 
         @Override
         protected boolean process() throws Exception
         {
             // Only return if EOF has previously been read and thus
             // a write done with EOF=true
             if (_eof)
             {
                 // Handle EOF
                 _in.close();
                 closed();
                 _channel.getByteBufferPool().release(_buffer);
                 return true;
             }
             
             // Read until buffer full or EOF
             int len=0;
             while (len<_buffer.capacity() && !_eof)
             {
                 int r=_in.read(_buffer.array(),_buffer.arrayOffset()+len,_buffer.capacity()-len);
                 if (r<0)
                     _eof=true;
                 else
                     len+=r;
             }
 
             // write what we have
             _buffer.position(0);
             _buffer.limit(len);
             _channel.write(_buffer,_eof,this);
             return false;
         }
 
         @Override
         public void failed(Throwable x)
         {
             super.failed(x);
             _channel.getByteBufferPool().release(_buffer);
             try
             {
                 _in.close();
             }
             catch (IOException e)
             {
                 LOG.ignore(e);
             }
         }
         
     }
 
     /* ------------------------------------------------------------ */
     /** An iterating callback that will take content from a 
      * ReadableByteChannel and write it to the {@link HttpChannel}.
      * A {@link ByteBuffer} of size {@link HttpOutput#getBufferSize()} is used that will be direct if
      * {@link HttpChannel#useDirectBuffers()} is true.
      * This callback is passed to the {@link HttpChannel#write(ByteBuffer, boolean, Callback)} to
      * be notified as each buffer is written and only once all the input is consumed will the 
      * wrapped {@link Callback#succeeded()} method be called. 
      */
     private class ReadableByteChannelWritingCB extends IteratingCallback
     {
         private final ReadableByteChannel _in;
         private final ByteBuffer _buffer;
         private boolean _eof;
         
         public ReadableByteChannelWritingCB(ReadableByteChannel in, Callback callback)
         {          
             super(callback);
             _in=in;
             _buffer = _channel.getByteBufferPool().acquire(getBufferSize(), _channel.useDirectBuffers());
         }
 
         @Override
         protected boolean process() throws Exception
         {
             // Only return if EOF has previously been read and thus
             // a write done with EOF=true
             if (_eof)
             {
                 _in.close();
                 closed();
                 _channel.getByteBufferPool().release(_buffer);
                 return true;
             }
             
             // Read from stream until buffer full or EOF
             _buffer.clear();
             while (_buffer.hasRemaining() && !_eof)
               _eof = (_in.read(_buffer)) <  0;
             
             // write what we have
             _buffer.flip();
             _channel.write(_buffer,_eof,this);
             return false;
         }
 
         @Override
         public void failed(Throwable x)
         {
             super.failed(x);
             _channel.getByteBufferPool().release(_buffer);
             try
             {
                 _in.close();
             }
             catch (IOException e)
             {
                 LOG.ignore(e);
             }
         }
     }
 }