Merlin brings nonblocking I/O to the Java platform

README

Sample code from the "Merlin brings nonblocking I/O to the Java platform" article. Really old, but usable.

http://www.ibm.com/developerworks/java/library/j-javaio

I done some refactoring. ;)

Client.java

package nonblocking.client;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;
import java.nio.charset.CharsetDecoder;

public class Client {

    private class MessageReceiver extends Thread {
        private SocketChannel clientSocketChannel;
        private boolean       shouldReceive;

        public MessageReceiver(String name, SocketChannel clientSocketChannel) {
            super(name);
            
            this.clientSocketChannel = clientSocketChannel;
        }
        
        public void startReceiving() {
            shouldReceive = true;
            start();
        }
        
        public void stopReceiving() {
            shouldReceive = false;
        }

        public void run() {
            System.out.println("Inside receive message");
            
            ByteBuffer byteBuffer = ByteBuffer.allocate(2048);
            
            try {
                while (shouldReceive) {
                    while (clientSocketChannel.read(byteBuffer) > 0) {
                        byteBuffer.flip();
                        
                        Charset        charset    = Charset.forName("us-ascii");
                        CharsetDecoder decoder    = charset.newDecoder();
                        CharBuffer     charBuffer = decoder.decode(byteBuffer);
                        String         result     = charBuffer.toString();
                        
                        System.out.println(result);
                        
                        byteBuffer.flip();
                    }
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
    
    ///
    
    private SocketChannel   clientSocketChannel;
    private MessageReceiver messageReceiver;

    public void makeConnection() {
        try {
            clientSocketChannel = setupClientSocketChannel();
        } catch (IOException e) {
            e.printStackTrace();
        }

        startReceivingMessage();
        
        while (sendMessage() != -1) {}
        
        stopReceivingMessage();
        
        try {
            Thread.sleep(5000);

            clientSocketChannel.close();
        } catch (IOException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    
    private SocketChannel setupClientSocketChannel() throws IOException {
        System.out.println("Inside setupClientSocketChannel");

        InetSocketAddress serverAddress = new InetSocketAddress(InetAddress.getLocalHost(), 4900);
        
        clientSocketChannel = SocketChannel.open();
        clientSocketChannel.connect(serverAddress);
        clientSocketChannel.configureBlocking(false);
        
        return clientSocketChannel;
    }

    private void startReceivingMessage() {
        messageReceiver = new MessageReceiver("MessageReceiver Thread", clientSocketChannel);
        messageReceiver.startReceiving();
    }

    private void stopReceivingMessage() {
        messageReceiver.stopReceiving();
    }

    private int sendMessage() {
        System.out.println("Inside SendMessage");
        
        BufferedReader input         = new BufferedReader(new InputStreamReader(System.in));
        String         message       = null;
        ByteBuffer     byteBuffer    = ByteBuffer.allocate(1024);
        int            numberOfBytes = 0;
        
        try {
            System.out.print(">> ");
            message = input.readLine();
            System.out.println("message is " + message);

            if (message.equals("quit") || message.equals("shutdown")) {
                System.out.println("Time to stop the client");
                
                return -1;
            }
            
            byteBuffer    = ByteBuffer.wrap(message.getBytes());
            numberOfBytes = clientSocketChannel.write(byteBuffer);
            System.out.println("numberOfBytes is " + numberOfBytes);
        } catch (IOException e) {
            e.printStackTrace();
        }
        
        System.out.println("Wrote " + numberOfBytes + " bytes to the server");
        
        return numberOfBytes;
    }

    public static void main(String args[]) {
        Client client = new Client();
        client.makeConnection();
    }
}

NonBlockingServer.java

package nonblocking.server;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.nio.channels.FileChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;
import java.nio.charset.CharsetDecoder;

public class NonBlockingServer {
    private ServerSocketChannel serverSocketChannel;
    private Selector            selector;

    public void startServer() throws IOException {
        System.out.println("Inside startServer");
        
        serverSocketChannel = setupServerSocketChannel();
        selector            = Selector.open();
        
        SelectionKey acceptKey = serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
        
        System.out.println("Abt to block on select()");

        while (acceptKey.selector().select() > 0) {
            for (SelectionKey readyKey : selector.selectedKeys()) {
                selector.selectedKeys().remove(readyKey);
                
                if (readyKey.isAcceptable()) {
                    System.out.println("Key is Acceptable");
                    
                    SocketChannel clientSocketChannel = (SocketChannel) serverSocketChannel.accept();
                    clientSocketChannel.configureBlocking(false);
                    clientSocketChannel.register(selector, SelectionKey.OP_READ | SelectionKey.OP_WRITE);
                } else if (readyKey.isReadable()) {
                    System.out.println("Key is Readable");
                    
                    SocketChannel clientSocketChannel = (SocketChannel) readyKey.channel();
                    String        message             = readMessage(clientSocketChannel);
                    
                    if (message.length() > 0) {
                        writeMessage(clientSocketChannel, message);
                    }
                } else if (readyKey.isWritable()) {
                    System.out.println("Key is Writable");
                    
                    SocketChannel clientSocketChannel = (SocketChannel) readyKey.channel();
                    String        message             = readMessage(clientSocketChannel);
                    
                    if (message.length() > 0) {
                        writeMessage(clientSocketChannel, message);
                    }
                }
            }
        }
    }

    private ServerSocketChannel setupServerSocketChannel() throws IOException {
        System.out.println("Inside setupServerSocketChannel");
        
        InetSocketAddress serverAddress = new InetSocketAddress(InetAddress.getLocalHost(), 4900);
        
        serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.configureBlocking(false);
        serverSocketChannel.socket().bind(serverAddress);
        
        return serverSocketChannel;
    }

    public String readMessage(SocketChannel clientSocketChannel) throws IOException {
        ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
    
        clientSocketChannel.read(byteBuffer);
        byteBuffer.flip();
        
        Charset        charset    = Charset.forName("us-ascii");
        CharsetDecoder decoder    = charset.newDecoder();
        CharBuffer     charBuffer = decoder.decode(byteBuffer);
        
        return charBuffer.toString();
    }

    public void writeMessage(SocketChannel clientSocketChannel, String message) {
        System.out.println("Inside the loop");

        if (message.equals("quit") || message.equals("shutdown")) {
            return;
        }
        
        File file = new File(message);
        
        try {
            RandomAccessFile randomAccessFile = new RandomAccessFile(file, "r");
            FileChannel      fileChannel      = randomAccessFile.getChannel();
            ByteBuffer       byteBuffer       = ByteBuffer.allocate(1024);
            
            fileChannel.read(byteBuffer);
            byteBuffer.flip();

            Charset        charset    = Charset.forName("us-ascii");
            CharsetDecoder decoder    = charset.newDecoder();
            CharBuffer     charBuffer = decoder.decode(byteBuffer);
            
            System.out.println("charBuffer = " + charBuffer.toString());
            
            byteBuffer = ByteBuffer.wrap((charBuffer.toString()).getBytes());
            int numberOfBytes = clientSocketChannel.write(byteBuffer);
            
            System.out.println("numberOfBytes = " + numberOfBytes);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static void main(String args[]) {
        NonBlockingServer nonBlockingServer = new NonBlockingServer();
        
        try {
            nonBlockingServer.startServer();
        } catch (IOException e) {
            e.printStackTrace();
            
            System.exit(-1);
        }
    }
}