[java] NIO server programming

[source="java"]package game.server;import java.io.*;import java.nio.*;import java.nio.channels.*;import java.util.*;import java.util.concurrent.*;/* * This class implements a multiplexed output system for multiple * socket channels. It takes requests to write ByteBuffers into * the given SocketChannels. It processes these requests in its own * thread. This is implemented internally using a ConcurrentLinkedQueue. */public class MultiplexedSocketWriter{	private static MultiplexedSocketWriter instance = null;	private SocketChannelOutputBufferPool bufferPool = SocketChannelOutputBufferPool.getInstance();	private ConcurrentLinkedQueue<SocketChannelOutputBuffer> outputQueue = new ConcurrentLinkedQueue<SocketChannelOutputBuffer>();	private boolean terminateThread = false;		/*	 * Start sending the message queue.	 */	private MultiplexedSocketWriter()	{				new Thread(new Runnable()		{			public void run()			{				while (!terminateThread)				{					while (!outputQueue.isEmpty())					{						Iterator<SocketChannelOutputBuffer> iter = outputQueue.iterator();						while (iter.hasNext())						{							SocketChannelOutputBuffer buffer = iter.next();							buffer.send();							if (buffer.sent())							{								bufferPool.release(buffer);								iter.remove();							}						}					}					// No messages to send, so wait until notified.					try					{						synchronized (MultiplexedSocketWriter.this)						{							wait();						}					}					catch (InterruptedException exc)					{					}				}			}		}).start();	}		/*	 * Get the only instance of MultiplexedSocketWriter	 */	public static MultiplexedSocketWriter getInstance()	{		if (instance == null)		{			instance = new MultiplexedSocketWriter();		}		return instance;	}		/*	 * Terminate this MultiplexedSocketWriter. It will never be restartable.	 * Once calling this method, no more messages will be able to be sent	 * using this writer for the remainder of program execution. This method	 * should be called if and only if the program must exit. Calling this	 * method will cause the internal thread to terminate once all pending	 * messages have been sent.	 */	public void terminate()	{		terminateThread = true;	}		/*	 * Send the ByteBuffer via the SocketChannel. This actually places	 * the ByteBuffer in a queue to be sent as soon as possible.	 */	public void send(SocketChannel channel, ByteBuffer buffer)	{		outputQueue.add(bufferPool.acquire(channel, buffer));		// Notify the queue consumer to wake up in case it was waiting.		synchronized (this)		{			notify();		}	}			/*	 * Mutable wrapper for a SocketChannel and a ByteBuffer to send	 * through it. Also contains a send method to signal the SocketChannel	 * to send the ByteBuffer.	 */	private static class SocketChannelOutputBuffer	{		private SocketChannel channel;		private ByteBuffer buffer;			/*		 * This constuctor does nothing in particular. It is only		 * of interest to the SocketChannelOutputBufferPool class.		 */		public SocketChannelOutputBuffer()		{			channel = null;			buffer = null;		}				/*		 * Set the SocketChannel and ByteBuffer to use.		 */		public SocketChannelOutputBuffer set(SocketChannel channel, ByteBuffer buffer)		{			this.channel = channel;			this.buffer = buffer;			buffer.flip();			return this;		}				/*		 * Send the ByteBuffer through the SocketChannel. This may or		 * may not send the entire ByteBuffer. Subsequent calls to this		 * method will attempt to send any remaining bytes.		 */		public void send()		{			try			{				channel.write(buffer);				if (!buffer.hasRemaining())				{					buffer.clear();				}			}			catch (IOException exc)			{				buffer.clear();		// This causes the next call to sent() to return true.				try				{					channel.close();	// Close this channel.				}				catch (IOException ignore)				{				}			}		}				/*		 * Returns the sending status of the ByteBuffer (true is send completed).		 */		public boolean sent()		{			return !buffer.hasRemaining();		}	}			/*	 * Pool of SocketChannelOutputBuffers. This saves newing an object each time	 * a message is to be sent. Let's hope LinkedList uses node pooling, otherwise	 * this is a waste of code!	 */	private static class SocketChannelOutputBufferPool	{		private static SocketChannelOutputBufferPool instance = null;		private static final int initialSize = 100;		private List<SocketChannelOutputBuffer> pool = new LinkedList<SocketChannelOutputBuffer>();				private SocketChannelOutputBufferPool()		{			allocate();		}				/*		 * Get the only allowable instance of this pool.		 */		public static SocketChannelOutputBufferPool getInstance()		{			if (instance == null)			{				instance = new SocketChannelOutputBufferPool();			}			return instance;		}				/*		 * Acquire a SocketChannelOutputBuffer from this pool		 * and set it to the given parameters.		 */				public SocketChannelOutputBuffer acquire(SocketChannel channel, ByteBuffer buffer)		{			if (pool.isEmpty())			{				allocate();			}			return pool.remove(0).set(channel, buffer);		}				/*		 * Release the SocketChannelOutputBuffer back into this pool.		 */		public void release(SocketChannelOutputBuffer buffer)		{			pool.add(buffer);		}				/*		 * Allocate some SocketChannelOutputBuffers in this pool.		 */		private void allocate()		{			for (int i = 0; i < initialSize; ++i)			{				pool.add(new SocketChannelOutputBuffer());			}		}	}}

import java.io.*;import java.nio.channels.*;import java.util.*;/* * This class implements a multiplexed output system for multiple * socket channels. It takes requests to write ByteBuffers into * the given SocketChannels. It processes these requests in its own * thread. This is implemented internally using a ConcurrentLinkedQueue. */public class MultiplexedSocketWriter{	private static MultiplexedSocketWriter instance = null;	private LinkedList<OutputWrapper> outputQueue = new LinkedList<OutputWrapper>();	private boolean terminateThread = false;	private Runnable thread = new OutputThread();		/*	 * Start sending the message queue.	 */	private MultiplexedSocketWriter()	{		new Thread(thread).start();	}		private class OutputThread implements Runnable	{		public void run()		{			while (!terminateThread)			{				// Don't allow the output queue to be modified just in case the output				// cannot be sent entirely. For any given channel, we don't want new				// output to be sent before old output.				synchronized (OutputThread.this)				{					while (outputQueue.isEmpty())					{						try						{							System.out.println("Waiting...");							wait();						}						catch (InterruptedException exc)						{						}					}					System.out.println("Got notified.");					// Take an output message from the queue					OutputWrapper output = outputQueue.removeFirst();					// Send the buffer.					try					{						output.buffer.get(output.channel);					}					catch (IOException exc)					{						// Not sure what to do yet. Perhaps just ignore silently.						// This could work because this indicates the channel is						// now closed, which can be handled in the selector loop.						System.out.println("Encountered IOException while draining output queue.");					}					// If the buffer has not been completely sent then put it back into the queue.					if (!output.buffer.isEmpty())					{						outputQueue.addLast(output);					}				}			}		}	}		/*	 * Get the only instance of MultiplexedSocketWriter	 */	public static MultiplexedSocketWriter getInstance()	{		if (instance == null)		{			instance = new MultiplexedSocketWriter();		}		return instance;	}		/*	 * Terminate this MultiplexedSocketWriter. It will never be restartable.	 * Once calling this method, no more messages will be able to be sent	 * using this writer for the remainder of program execution. This method	 * should be called if and only if the program must exit. Calling this	 * method will cause the internal thread to terminate once all pending	 * messages have been sent.	 */	public void terminate()	{		terminateThread = true;	}		/*	 * Send the SocketChannelOutputBuffer via the SocketChannel. This actually places	 * the SocketChannelOutputBuffer in a queue to be sent as soon as possible.	 * If the buffer is already in the queue, this method returns true (success).	 * If the queue is full, the buffer is not added to the queue and this method	 * returns false.	 */	public void send(SocketChannel channel, SocketChannelOutputBuffer buffer)	{		synchronized (thread)		{			System.out.println("Sending...");			outputQueue.addLast(new OutputWrapper(channel, buffer));			thread.notify();		}	}		private static class OutputWrapper	{		public SocketChannel channel;		public SocketChannelOutputBuffer buffer;				public OutputWrapper(SocketChannel channel, SocketChannelOutputBuffer buffer)		{			this.channel = channel;			this.buffer = buffer;		}	}}

import java.io.*;import java.nio.channels.*;public interface SocketChannelInputBuffer{	public int put(SocketChannel channel) throws IOException;	public SocketChannelInputBuffer put(byte b);	public SocketChannelInputBuffer put(byte[] src);	public SocketChannelBuffer put(SocketChannelBuffer src);	public boolean isEmpty();}

import java.io.*;import java.nio.channels.*;public interface SocketChannelOutputBuffer{	public int get(SocketChannel channel) throws IOException;	public byte get();	public SocketChannelOutputBuffer get(byte[] dst);	public boolean isEmpty();}

import java.nio.*;import java.nio.channels.*;import java.io.*;public class SocketChannelBuffer implements SocketChannelInputBuffer, SocketChannelOutputBuffer{	private static final int STATE_READ = 1;	private static final int STATE_WRITE = 2;	private ByteBuffer internalBuffer;	private int size;	private int byteCount;	private int state;		public SocketChannelBuffer(int size, boolean direct)	{		this.size = size;		internalBuffer = direct ? ByteBuffer.allocate(size) : ByteBuffer.allocateDirect(size);		internalBuffer.clear();		byteCount = 0;		state = STATE_READ;	}		public synchronized int put(SocketChannel channel) throws IOException	{		if (byteCount == size)		{			return 0;		}		setRead();		int bytesRead = channel.read(internalBuffer);		// We don't want to update byteCount if bytesRead is -1.		// May as well also catch btyesRead == 0 for efficiency.		if (bytesRead > 0)		{			byteCount += bytesRead;		}		return bytesRead;	}		public synchronized int get(SocketChannel channel) throws IOException	{		if (byteCount == 0)		{			return 0;		}		setWrite();		int bytesWritten = channel.write(internalBuffer);		byteCount -= bytesWritten;		return bytesWritten;	}		public synchronized SocketChannelBuffer put(byte b)	{		if (byteCount == size)		{			throw new BufferOverflowException();		}		setRead();		internalBuffer.put(b);		++byteCount;		return this;	}		public synchronized byte get()	{		if (byteCount == 0)		{			throw new BufferUnderflowException();		}		setWrite();		--byteCount;		return internalBuffer.get();	}		public synchronized SocketChannelBuffer put(byte[] src)	{		if (byteCount + src.length > size)		{			throw new BufferOverflowException();		}		setRead();		internalBuffer.put(src);		byteCount += src.length;		return this;	}		public synchronized SocketChannelBuffer get(byte[] dst)	{		if (byteCount < dst.length)		{			throw new BufferUnderflowException();		}		setWrite();		internalBuffer.get(dst);		byteCount -= dst.length;		return this;	}		public synchronized SocketChannelBuffer put(SocketChannelBuffer src)	{		while (!src.isEmpty() && byteCount < size)		{			this.put(src.get());		}		return this;	}		public synchronized boolean isEmpty()	{		return byteCount == 0;	}		private void setRead()	{		switch (state)		{		case STATE_READ:			return;		case STATE_WRITE:			if (byteCount == 0)			{				internalBuffer.clear();			}			else // Still some bytes remaining.			{				System.out.println("Compacting.");				internalBuffer.compact();			}			state = STATE_READ;		}	}		private void setWrite()	{		switch (state)		{		case STATE_READ:			internalBuffer.flip();			state = STATE_WRITE;			return;		case STATE_WRITE:			return;					}	}}