Simple non-blocking server

Question

To write an instant messenger program, I am trying to compose a simple server class that will run in its thread.

What should the server do

• accept connections from / connect to other server instances and bind selection keys for connections in Map<Integer, SelectionKey> keys with an identifier so that the message flow can access the connections by identifier
• read / write to connections
• keep incoming messages in the queue
• message flow can
  • selection of incoming messages
  • sent messages: send_message(int id, String msg)

My current approach is mainly based on this example: A simple non-blocking echo server with Java nio .
I also used the Use of a selector to manage non-blocking sockets and real pages to learn about non-blocking sockets and selectors.

Current code

• EJP Offers Implemented
• minor changes

 package snserver; /* imports */ //class SNServer (Simple non-blocking Server) public class SNServer extends Thread { private int port; private Selector selector; private ConcurrentMap<Integer, SelectionKey> keys; // ID -> associated key private ConcurrentMap<SocketChannel,List<byte[]>> dataMap_out; ConcurrentLinkedQueue<String> in_msg; //incoming messages to be fetched by messenger thread public SNServer(int port) { this.port = port; dataMap_out = new ConcurrentHashMap<SocketChannel, List<byte[]>>(); keys = new ConcurrentHashMap<Integer, SelectionKey>(); } public void start_server() throws IOException { // create selector and channel this.selector = Selector.open(); ServerSocketChannel serverChannel = ServerSocketChannel.open(); serverChannel.configureBlocking(false); // bind to port InetSocketAddress listenAddr = new InetSocketAddress((InetAddress)null, this.port); serverChannel.socket().bind(listenAddr); serverChannel.register(this.selector, SelectionKey.OP_ACCEPT); log("Echo server ready. Ctrl-C to stop."); // processing while (true) { // wait for events this.selector.select(); // wakeup to work on selected keys Iterator keys = this.selector.selectedKeys().iterator(); while (keys.hasNext()) { SelectionKey key = (SelectionKey) keys.next(); // this is necessary to prevent the same key from coming up // again the next time around. keys.remove(); if (! key.isValid()) { continue; } if (key.isAcceptable()) { this.accept(key); } else if (key.isReadable()) { this.read(key); } else if (key.isWritable()) { this.write(key); } else if(key.isConnectable()) { this.connect(key); } } } } private void accept(SelectionKey key) throws IOException { ServerSocketChannel serverChannel = (ServerSocketChannel) key.channel(); SocketChannel channel = serverChannel.accept(); channel.configureBlocking(false); send_message(key, "Welcome."); //DEBUG Socket socket = channel.socket(); SocketAddress remoteAddr = socket.getRemoteSocketAddress(); log("Connected to: " + remoteAddr); // register channel with selector for further IO dataMap_out.put(channel, new ArrayList<byte[]>()); channel.register(this.selector, SelectionKey.OP_READ); //store key in 'keys' to be accessable by ID from messenger thread //TODO first get ID keys.put(0, key); } //TODO verify, test public void init_connect(String addr, int port){ try { SocketChannel channel = createSocketChannel(addr, port); channel.register(this.selector, channel.validOps()/*, SelectionKey.OP_?*/); } catch (IOException e) { //TODO handle } } //TODO verify, test private void connect(SelectionKey key) { SocketChannel channel = (SocketChannel) key.channel(); try { channel.finishConnect(); //try to finish connection - if 'false' is returned keep 'OP_CONNECT' registered //store key in 'keys' to be accessable by ID from messenger thread //TODO first get ID keys.put(0, key); } catch (IOException e0) { try { //TODO handle ok? channel.close(); } catch (IOException e1) { //TODO handle } } } private void read(SelectionKey key) throws IOException { SocketChannel channel = (SocketChannel) key.channel(); ByteBuffer buffer = ByteBuffer.allocate(8192); int numRead = -1; try { numRead = channel.read(buffer); } catch (IOException e) { e.printStackTrace(); } if (numRead == -1) { this.dataMap_out.remove(channel); Socket socket = channel.socket(); SocketAddress remoteAddr = socket.getRemoteSocketAddress(); log("Connection closed by client: " + remoteAddr); //TODO handle channel.close(); return; } byte[] data = new byte[numRead]; System.arraycopy(buffer.array(), 0, data, 0, numRead); in_msg.add(new String(data, "utf-8")); } private void write(SelectionKey key) throws IOException { SocketChannel channel = (SocketChannel) key.channel(); List<byte[]> pendingData = this.dataMap_out.get(channel); Iterator<byte[]> items = pendingData.iterator(); while (items.hasNext()) { byte[] item = items.next(); items.remove(); //TODO is this correct? -> re-doing write in loop with same buffer object ByteBuffer buffer = ByteBuffer.wrap(item); int bytes_to_write = buffer.capacity(); while (bytes_to_write > 0) { bytes_to_write -= channel.write(buffer); } } key.interestOps(SelectionKey.OP_READ); } public void queue_data(SelectionKey key, byte[] data) { SocketChannel channel = (SocketChannel) key.channel(); List<byte[]> pendingData = this.dataMap_out.get(channel); key.interestOps(SelectionKey.OP_WRITE); pendingData.add(data); } public void send_message(int id, String msg) { SelectionKey key = keys.get(id); if (key != null) send_message(key, msg); //else //TODO handle } public void send_message(SelectionKey key, String msg) { try { queue_data(key, msg.getBytes("utf-8")); } catch (UnsupportedEncodingException ex) { //is not thrown: utf-8 is always defined } } public String get_message() { return in_msg.poll(); } private static void log(String s) { System.out.println(s); } @Override public void run() { try { start_server(); } catch (IOException e) { System.out.println("IOException: " + e); //TODO handle exception } } // Creates a non-blocking socket channel for the specified host name and port. // connect() is called on the new channel before it is returned. public static SocketChannel createSocketChannel(String hostName, int port) throws IOException { // Create a non-blocking socket channel SocketChannel sChannel = SocketChannel.open(); sChannel.configureBlocking(false); // Send a connection request to the server; this method is non-blocking sChannel.connect(new InetSocketAddress(hostName, port)); return sChannel; } } 

My question is: Is the above code correct and good, or what should I change? How to properly fulfill the requirements mentioned above? Also pay attention to my "TODO".

Thanks for the help!

Answer 1

There are several issues here.

• You do not check the result of write (). He can return anything from scratch. You may need to re-do this several times.

• If finishConnect () returns false, this is not an error, it is not finished yet, so just leave the registered OP_CONNECT and wait for it (again). The only validOps () for SocketChannel that you just created through SocketChannel.open () is OP_CONNECT. If finishConnect () throws an exception, this is an error and you must close the channel.

• Closing a channel cancels the key; you do not need to cancel it.

• Generally, you should use null as a local InetAddress when binding.