/*

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

 		}

 	}

 }