package simple;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Created by huangyinhuang on 2/13/2018.
 * a simplified demo of AsynThreadSelectorServer
 */
public class SimpleAsyncThreadSelectorServer {

    public static void main(String[] args) throws Exception {
        // create a thread pool
        ExecutorService threadPool = Executors.newFixedThreadPool(3);

        // please note: configure channel blocking state as false
        System.out.println("start the server.");
        ServerSocketChannel server = ServerSocketChannel.open();
        server.configureBlocking(false);

        ServerSocket serverSocket = server.socket();
        serverSocket.bind(new InetSocketAddress(9000));

        // register the channel
        Selector selector = Selector.open();
        server.register(selector, SelectionKey.OP_ACCEPT);

        // check new connection's events at intervals
        while (true) {
            Thread.sleep(500);

            // please note: selector.select() will block current thread
            if (selector.select() == 0) {
                continue;
            }

            Set<SelectionKey> selectedKeys = selector.selectedKeys();
            Iterator<SelectionKey> keyIterator = selectedKeys.iterator();
            while (keyIterator.hasNext()) {
                SelectionKey key = keyIterator.next();

                if (key.isAcceptable()) {
                    // a connection was accepted by a ServerSocketChannel.
                    ServerSocketChannel srv = (ServerSocketChannel) key.channel();
                    SocketChannel socket = srv.accept();
                    socket.configureBlocking(false);
                    socket.register(selector, SelectionKey.OP_READ);
                } else if (key.isReadable()) {
                    // a channel is ready for reading
                    SocketChannel channel = (SocketChannel) key.channel();
                    // handle upcoming new channel message by thread
                    threadPool.submit(new HandlerThread(channel));
                }

                keyIterator.remove();
            }
        }

    }

    public static class HandlerThread implements Runnable {

        private SocketChannel channel;

        public HandlerThread(SocketChannel channel) {
            this.channel = channel;
        }

        @Override
        public void run() {
            System.out.println("try to read the message from channel.");
            ByteBuffer buffer = ByteBuffer.allocate(64);

            try {
                if (channel.read(buffer) != -1) {
                    buffer.flip();

                    if (buffer.hasRemaining()) {
                        byte[] msgBytes = new byte[buffer.remaining()];
                        buffer.get(msgBytes);
                        System.out.println(new String(msgBytes).trim());
                    }

                    buffer.clear();
                } else {
                    System.out.println("Client has closed the connection, shutdown the channel.");
                    channel.close();
                }
            } catch (IOException e) {
                System.out.println("An exception is thrown when trying to read/close channel. Details: " + e.getMessage());
                try {
                    channel.close();
                } catch (IOException e1) {
                    e1.printStackTrace();
                }
            }
        }

    }

}