/*

  *   SONEWS News Server

  *   see AUTHORS for the list of contributors

  *

  *   This program is free software: you can redistribute it and/or modify

  *   it under the terms of the GNU General Public License as published by

  *   the Free Software Foundation, either version 3 of the License, or

  *   (at your option) any later version.

  *

  *   This program is distributed in the hope that it will be useful,

  *   but WITHOUT ANY WARRANTY; without even the implied warranty of

  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

  *   GNU General Public License for more details.

  *

  *   You should have received a copy of the GNU General Public License

  *   along with this program.  If not, see <http://www.gnu.org/licenses/>.

  */

 package org.sonews.daemon;

 import java.nio.ByteBuffer;

 import java.nio.channels.ClosedChannelException;

 import java.util.ArrayList;

 import java.util.List;

 /**

  * Class holding ByteBuffers for SocketChannels/NNTPConnection.

  * Due to the complex nature of AIO/NIO we must properly handle the line 

  * buffers for the input and output of the SocketChannels.

  * @author Christian Lins

  * @since sonews/0.5.0

  */

 public class ChannelLineBuffers 

 {

   /**

    * Size of one small buffer; 

    * per default this is 512 bytes to fit one standard line.

    */

   public static final int BUFFER_SIZE = 512;

   private static int maxCachedBuffers = 2048; // Cached buffers maximum

   private static final List<ByteBuffer> freeSmallBuffers

     = new ArrayList<ByteBuffer>(maxCachedBuffers);

   /**

    * Allocates a predefined number of direct ByteBuffers (allocated via

    * ByteBuffer.allocateDirect()). This method is Thread-safe, but should only

    * called at startup.

    */

   public static void allocateDirect()

   {

     synchronized(freeSmallBuffers)

     {

       for(int n = 0; n < maxCachedBuffers; n++)

       {

         ByteBuffer buffer = ByteBuffer.allocateDirect(BUFFER_SIZE);

         freeSmallBuffers.add(buffer);

       }

     }

   }

   private ByteBuffer       inputBuffer   = newLineBuffer();

   private List<ByteBuffer> outputBuffers = new ArrayList<ByteBuffer>();

   /**

    * Add the given ByteBuffer to the list of buffers to be send to the client.

    * This method is Thread-safe.

    * @param buffer

    * @throws java.nio.channels.ClosedChannelException If the client channel was

    * already closed.

    */

   public void addOutputBuffer(ByteBuffer buffer)

     throws ClosedChannelException

   {

     if(outputBuffers == null)

     {

       throw new ClosedChannelException();

     }

     synchronized(outputBuffers)

     {

       outputBuffers.add(buffer);

     }

   }

   /**

    * Currently a channel has only one input buffer. This *may* be a bottleneck

    * and should investigated in the future.

    * @param channel

    * @return The input buffer associated with given channel.

    */

   public ByteBuffer getInputBuffer()

   {

     return inputBuffer;

   }

   /**

    * Returns the current output buffer for writing(!) to SocketChannel.

    * @param channel

    * @return The next input buffer that contains unprocessed data or null

    * if the connection was closed or there are no more unprocessed buffers.

    */

   public ByteBuffer getOutputBuffer()

   {

     synchronized(outputBuffers)

     {

       if(outputBuffers == null || outputBuffers.isEmpty())

       {

         return null;

       }

       else

       {

         ByteBuffer buffer = outputBuffers.get(0);

         if(buffer.remaining() == 0)

         {

           outputBuffers.remove(0);

           // Add old buffers to the list of free buffers

           recycleBuffer(buffer);

           buffer = getOutputBuffer();

         }

         return buffer;

       }

     }

   }

   /**

    * @return false if there are output buffers pending to be written to the

    * client.

    */

   boolean isOutputBufferEmpty()

   {

     synchronized(outputBuffers)

     {

       return outputBuffers.isEmpty();

     }

   }

   /**

    * Goes through the input buffer of the given channel and searches

    * for next line terminator. If a '\n' is found, the bytes up to the

    * line terminator are returned as array of bytes (the line terminator

    * is omitted). If none is found the method returns null.

    * @param channel

    * @return A ByteBuffer wrapping the line.

    */

   ByteBuffer nextInputLine()

   {

     if(inputBuffer == null)

     {

       return null;

     }

     synchronized(inputBuffer)

     {

       ByteBuffer buffer = inputBuffer;

       // Mark the current write position

       int mark = buffer.position();

       // Set position to 0 and limit to current position

       buffer.flip();

       ByteBuffer lineBuffer = newLineBuffer();

       while (buffer.position() < buffer.limit())

       {

         byte b = buffer.get();

         if (b == 10) // '\n'

         {

           // The bytes between the buffer's current position and its limit, 

           // if any, are copied to the beginning of the buffer. That is, the 

           // byte at index p = position() is copied to index zero, the byte at 

           // index p + 1 is copied to index one, and so forth until the byte 

           // at index limit() - 1 is copied to index n = limit() - 1 - p. 

           // The buffer's position is then set to n+1 and its limit is set to 

           // its capacity.

           buffer.compact();

           lineBuffer.flip(); // limit to position, position to 0

           return lineBuffer;

         }

         else

         {

           lineBuffer.put(b);

         }

       }

       buffer.limit(BUFFER_SIZE);

       buffer.position(mark);

       if(buffer.hasRemaining())

       {

         return null;

       }

       else

       {

         // In the first 512 was no newline found, so the input is not standard

         // compliant. We return the current buffer as new line and add a space

         // to the beginning of the next line which corrects some overlong header

         // lines.

         inputBuffer = newLineBuffer();

         inputBuffer.put((byte)' ');

         buffer.flip();

         return buffer;

       }

     }

   }

   /**

    * Returns a at least 512 bytes long ByteBuffer ready for usage.

    * The method first try to reuse an already allocated (cached) buffer but

    * if that fails returns a newly allocated direct buffer.

    * Use recycleBuffer() method when you do not longer use the allocated buffer.

    */

   static ByteBuffer newLineBuffer()

   {

     ByteBuffer buf = null;

     synchronized(freeSmallBuffers)

     {

       if(!freeSmallBuffers.isEmpty())

       {

         buf = freeSmallBuffers.remove(0);

       }

     }

     if(buf == null)

     {

       // Allocate a non-direct buffer

       buf = ByteBuffer.allocate(BUFFER_SIZE);

     }

     assert buf.position() == 0;

     assert buf.limit() >= BUFFER_SIZE;

     return buf;

   }

   /**

    * Adds the given buffer to the list of free buffers if it is a valuable

    * direct allocated buffer.

    * @param buffer

    */

   public static void recycleBuffer(ByteBuffer buffer)

   {

     assert buffer != null;

     if(buffer.isDirect())

     {

       assert buffer.capacity() >= BUFFER_SIZE;

       // Add old buffers to the list of free buffers

       synchronized(freeSmallBuffers)

       {

         buffer.clear(); // Set position to 0 and limit to capacity

         freeSmallBuffers.add(buffer);

       }

     } // if(buffer.isDirect())

   }

   /**

    * Recycles all buffers of this ChannelLineBuffers object.

    */

   public void recycleBuffers()

   {

     synchronized(inputBuffer)

     {

       recycleBuffer(inputBuffer);

       this.inputBuffer = null;

     }

     synchronized(outputBuffers)

     {

       for(ByteBuffer buf : outputBuffers)

       {

         recycleBuffer(buf);

       }

       outputBuffers = null;

     }

   }

 }