/*

 *   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;

    }

  }

}