Java Nio Usage Demo & A example of use selectableChannel

Primary purpose of this blog is to use http://kylinsoong.iteye.com/blog/1250681 Listed API, complete a demo application: this application has two part, one is the Server, the other is Client, Client send a message to the Server Side, the Server response the recieved message to Client. as following picture depicted:

 This is Sever side logical sequence, Sever Side Open a ServerSocketChannel, use no-blocking mode, set resueAddress as true, bind IP address "localhost" and port number "8000";

The serverSocketChannel register a OP_ACCEPT Event to a opened Selector, use a look listen-in the OP_ACCEPT Event, while OP_ACCEPT Event occuring, the server retrieve the selection key from key set and deal with the key according to the key's attribute:

if it's a isAcceptable Event, the server side get the Socketchannel through Channel, also set no-blocking is true, register the OP_READ and OP_WRITE Event to the same Selector, came to loop again.

if it's a isReadable Event, Read th e data from channel, output to the console;

if it's a idWriteable Event, Write the echo message to Client Side through the channel.

 The Client Side is much clearer. Firstly, the client side open a SocketChannel, connect to the Server Side through Server registered IP addrerss and Port number, the same as Server Side set the bolocking is false, open a Selector, register a OP_READ and OP_WRITE Event to this selector, the came to the loop circle, listen-in the OP_READ and _OP_WRITE Event, while the registered event happened, retrieved the info from the channel and handle the Event, if it's isReadable, output it from console, in other, send the data to Server Side through the SocketChannel;

The Server Side Source Code

public class TestEchoServer {
	
	private static final Logger logger = Logger.getLogger(TestEchoServer.class);
	
	private Selector selector = null;
	
	private ServerSocketChannel serverSocketChannel = null;
	
	private int port = 8000;
	
	private Charset charset=Charset.forName("UTF-8");

	public TestEchoServer()throws IOException{
		selector = Selector.open();
		serverSocketChannel= ServerSocketChannel.open();
		serverSocketChannel.socket().setReuseAddress(true);
		serverSocketChannel.configureBlocking(false);
		serverSocketChannel.socket().bind(new InetSocketAddress(port));
		logger.info("Server is start[" + new InetSocketAddress(port).getAddress() + ":" + port + "]");
	}

	public void service() throws IOException{
		
		logger.info("ServerSocketChannel register [" + SelectionKey.OP_ACCEPT + "] to selector");
		serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT );
		while (selector.select() > 0) {
//			logger.debug("Select.select() value: " + selector.select());
			Set readyKeys = selector.selectedKeys();
			Iterator it = readyKeys.iterator();
			while (it.hasNext()) {
				SelectionKey key = null;
				try {
					key = (SelectionKey) it.next();
					it.remove();

					if (key.isAcceptable()) {
						logger.info("Selection Key isAcceptable: " + key.isAcceptable());
						ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
						SocketChannel socketChannel = (SocketChannel) ssc.accept();
						logger.info("Recieved Client Connection:" + socketChannel.socket().getInetAddress() + ":" + socketChannel.socket().getPort());
						socketChannel.configureBlocking(false);
						
						ByteBuffer buffer = ByteBuffer.allocate(1024);
						socketChannel.register(selector, SelectionKey.OP_READ | SelectionKey.OP_WRITE, buffer);
						logger.info("SocketChannel register [" + SelectionKey.OP_READ  + "] and [" + SelectionKey.OP_WRITE + "] to selector");
					}
					
					if (key.isReadable()) {
						logger.info("Selection Key isReadable");
						receive(key);
					}
					
					if (key.isWritable()) {
//						logger.info("Selection Key isWritable");
						send(key);
					}
				} catch (IOException e) {
					e.printStackTrace();
					try {
						if (key != null) {
							key.cancel();
							key.channel().close();
						}
					} catch (Exception ex) {
						e.printStackTrace();
					}
				}
			}
		}
	}

	public void send(SelectionKey key) throws IOException {
//		logger.info("send");
		ByteBuffer buffer = (ByteBuffer) key.attachment();
		SocketChannel socketChannel = (SocketChannel) key.channel();
		buffer.flip(); 
		String data = decode(buffer);
		if (data.indexOf("\r\n") == -1){
			return;
		}
		String outputData = data.substring(0, data.indexOf("\n") + 1);
		logger.info("outputData: " + outputData);
		ByteBuffer outputBuffer = encode("echo:" + outputData);
		while (outputBuffer.hasRemaining()){
			socketChannel.write(outputBuffer);
		}

		ByteBuffer temp = encode(outputData);
		buffer.position(temp.limit());
		buffer.compact();

		if (outputData.equals("bye\r\n")) {
			key.cancel();
			socketChannel.close();
			logger.info("Close Client Connection");
		}
	}

	public void receive(SelectionKey key) throws IOException {
		
		logger.info("receive");
		
		ByteBuffer buffer = (ByteBuffer) key.attachment();

		SocketChannel socketChannel = (SocketChannel) key.channel();
		ByteBuffer readBuff = ByteBuffer.allocate(32 * 1024);
		socketChannel.read(readBuff);
		readBuff.flip();

		buffer.limit(buffer.capacity());
		buffer.put(readBuff);
		logger.info("Recieved data: " + decode(buffer));
	}

	public String decode(ByteBuffer buffer) { 
		CharBuffer charBuffer = charset.decode(buffer);
		return charBuffer.toString();
	}

	public ByteBuffer encode(String str) { 
		return charset.encode(str);
	}

	public static void main(String args[]) throws Exception {
		TestEchoServer server = new TestEchoServer();
		server.service();
	}
}

 

The Client Side Source Code:

public class TestEchoClient {

	private SocketChannel socketChannel = null;
	
	private ByteBuffer sendBuffer = ByteBuffer.allocate(32 * 1024);
	
	private ByteBuffer receiveBuffer = ByteBuffer.allocate(32 * 1024);
	
	private Charset charset = Charset.forName("UTF-8");
	
	private Selector selector;
	
	private boolean isSend = false;
	
	public TestEchoClient() throws IOException {
		socketChannel = SocketChannel.open();
		InetAddress ia = InetAddress.getLocalHost();
		InetSocketAddress isa = new InetSocketAddress(ia, 8000);
		socketChannel.connect(isa);
		socketChannel.configureBlocking(false);
		log("Client: connection establish[" + isa.getHostName() + ":" + isa.getPort() + "]", false);
		selector = Selector.open();
	}

	public static void main(String args[]) throws IOException {
		final TestEchoClient client = new TestEchoClient();
		Thread receiver = new Thread() {
			public void run() {
				client.receiveFromUser();
			}
		};

		receiver.start();
		client.talk();
	}

	public void receiveFromUser() {
		try {
			BufferedReader localReader = new BufferedReader(new InputStreamReader(System.in));
			String msg = null;
			while ((msg = localReader.readLine()) != null) {
				log("Read From System Console: " + msg, false);
				isSend = true;
				log("Command line thread set isSend: " + isSend, false);
				synchronized (sendBuffer) {
					sendBuffer.put(encode(msg + "\r\n"));
				}
				if (msg.equals("bye")){
					log("Client Exit", false);
					break;
				}
			}
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	public void talk() throws IOException {
		log("SocketChannel register [" + SelectionKey.OP_READ  + "] and [" + SelectionKey.OP_WRITE + "] to selector", false);
		socketChannel.register(selector, SelectionKey.OP_READ | SelectionKey.OP_WRITE);
		while (selector.select() > 0) {
			Set readyKeys = selector.selectedKeys();
			Iterator it = readyKeys.iterator();
			while (it.hasNext()) {
				SelectionKey key = null;
				try {
					key = (SelectionKey) it.next();
					it.remove();

					if (key.isReadable()) {
						receive(key);
					}
					if (key.isWritable()) {
						send(key);
					}
				} catch (IOException e) {
					e.printStackTrace();
					try {
						if (key != null) {
							key.cancel();
							key.channel().close();
						}
					} catch (Exception ex) {
						e.printStackTrace();
					}
				}
			}
		}
	}

	public void send(SelectionKey key) throws IOException {
		if(!isSend) {
			return;
		}
		SocketChannel socketChannel = (SocketChannel) key.channel();
		synchronized (sendBuffer) {
			sendBuffer.flip(); 
			socketChannel.write(sendBuffer);
			sendBuffer.compact();
		}
		isSend = false;
		log("Send method set isSend: " + isSend, false );
	}

	public void receive(SelectionKey key) throws IOException {
		SocketChannel socketChannel = (SocketChannel) key.channel();
		socketChannel.read(receiveBuffer);
		receiveBuffer.flip();
		String receiveData = decode(receiveBuffer);

		if (receiveData.indexOf("\n") == -1){
			return;
		}

		String outputData = receiveData.substring(0, receiveData.indexOf("\n") + 1);
		log("Recieve Data: " + outputData, false);
		if (outputData.equals("echo:bye\r\n")) {
			key.cancel();
			socketChannel.close();
			System.out.println("Exit");
			selector.close();
			System.exit(0);
		}

		ByteBuffer temp = encode(outputData);
		receiveBuffer.position(temp.limit());
		receiveBuffer.compact();
	}

	public String decode(ByteBuffer buffer) {
		CharBuffer charBuffer = charset.decode(buffer);
		return charBuffer.toString();
	}

	public ByteBuffer encode(String str) { 
		return charset.encode(str);
	}
	
	public void log(String msg, boolean err) {
		String type = err ? "[WARN]" : "[INFO]";
		msg = new Date() + " --> " + type + " " + msg;
		System.out.println(msg);
	}
}