/*

 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.

 *

 * This code is free software; you can redistribute it and/or modify it

 * under the terms of the GNU General Public License version 2 only, as

 * published by the Free Software Foundation.

 *

 * This code is distributed in the hope that it will be useful, but WITHOUT

 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or

 * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License

 * version 2 for more details (a copy is included in the LICENSE file that

 * accompanied this code).

 *

 * You should have received a copy of the GNU General Public License version

 * 2 along with this work; if not, write to the Free Software Foundation,

 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.

 *

 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA

 * or visit www.oracle.com if you need additional information or have any

 * questions.

 */

/* @test

 * @summary Test selectors and socketchannels

 * @library ..

 */

import java.io.*;

import java.net.*;

import java.nio.*;

import java.nio.channels.*;

import java.nio.channels.spi.SelectorProvider;

import java.util.*;

public class SelectorTest {

    private static List clientList = new LinkedList();

    private static Random rnd = new Random();

    public static int NUM_CLIENTS = 30;

    public static int TEST_PORT = 31452;

    static PrintStream log = System.err;

    private static int FINISH_TIME = 30000;

    /*

     * Usage note

     *

     * java SelectorTest [server] [client <host>] [<port>]

     *

     * No arguments runs both client and server in separate threads

     * using the default port of 31452.

     *

     * client runs the client on this machine and connects to server

     * at the given IP address.

     *

     * server runs the server on localhost.

     */

    public static void main(String[] args) throws Exception {

        if (args.length == 0) {

            Server server = new Server(0);

            server.start();

            try {

                Thread.sleep(1000);

            } catch (InterruptedException e) { }

            InetSocketAddress isa

                = new InetSocketAddress(InetAddress.getLocalHost(), server.port());

            Client client = new Client(isa);

            client.start();

            if ((server.finish(FINISH_TIME) & client.finish(FINISH_TIME)) == 0)

                throw new Exception("Failure");

            log.println();

        } else if (args[0].equals("server")) {

            if (args.length > 1)

                TEST_PORT = Integer.parseInt(args[1]);

            Server server = new Server(TEST_PORT);

            server.start();

            if (server.finish(FINISH_TIME) == 0)

                throw new Exception("Failure");

            log.println();

        } else if (args[0].equals("client")) {

            if (args.length < 2) {

                log.println("No host specified: terminating.");

                return;

            }

            String ip = args[1];

            if (args.length > 2)

                TEST_PORT = Integer.parseInt(args[2]);

            InetAddress ia = InetAddress.getByName(ip);

            InetSocketAddress isa = new InetSocketAddress(ia, TEST_PORT);

            Client client = new Client(isa);

            client.start();

            if (client.finish(FINISH_TIME) == 0)

                throw new Exception("Failure");

            log.println();

        } else {

            System.out.println("Usage note:");

            System.out.println("java SelectorTest [server] [client <host>] [<port>]");

            System.out.println("No arguments runs both client and server in separate threads using the default port of 31452.");

            System.out.println("client runs the client on this machine and connects to the server specified.");

            System.out.println("server runs the server on localhost.");

        }

    }

    static class Client extends TestThread {

        InetSocketAddress isa;

        Client(InetSocketAddress isa) {

            super("Client", SelectorTest.log);

            this.isa = isa;

        }

        public void go() throws Exception {

            log.println("starting client...");

            for (int i=0; i<NUM_CLIENTS; i++)

                clientList.add(new RemoteEntity(i, isa, log));

            Collections.shuffle(clientList);

            log.println("created "+NUM_CLIENTS+" clients");

            do {

                for (Iterator i = clientList.iterator(); i.hasNext(); ) {

                    RemoteEntity re = (RemoteEntity) i.next();

                    if (re.cycle()) {

                        i.remove();

                    }

                }

                Collections.shuffle(clientList);

            } while (clientList.size() > 0);

        }

    }

    static class Server extends TestThread {

        private final ServerSocketChannel ssc;

        private List socketList = new ArrayList();

        private ServerSocket ss;

        private int connectionsAccepted = 0;

        private Selector pollSelector;

        private Selector acceptSelector;

        private Set pkeys;

        private Set pskeys;

        Server(int port) throws IOException {

            super("Server", SelectorTest.log);

            this.ssc = ServerSocketChannel.open().bind(new InetSocketAddress(port));

        }

        int port() {

            return ssc.socket().getLocalPort();

        }

        public void go() throws Exception {

            log.println("starting server...");

            acceptSelector = SelectorProvider.provider().openSelector();

            pollSelector = SelectorProvider.provider().openSelector();

            pkeys = pollSelector.keys();

            pskeys = pollSelector.selectedKeys();

            Set readyKeys = acceptSelector.selectedKeys();

            RequestHandler rh = new RequestHandler(pollSelector, log);

            Thread requestThread = new Thread(rh);

            requestThread.start();

            ssc.configureBlocking(false);

            SelectionKey acceptKey = ssc.register(acceptSelector,

                                                  SelectionKey.OP_ACCEPT);

            while(connectionsAccepted < SelectorTest.NUM_CLIENTS) {

                int keysAdded = acceptSelector.select(100);

                if (keysAdded > 0) {

                    Iterator i = readyKeys.iterator();

                    while(i.hasNext()) {

                        SelectionKey sk = (SelectionKey)i.next();

                        i.remove();

                        ServerSocketChannel nextReady =

                            (ServerSocketChannel)sk.channel();

                        SocketChannel sc = nextReady.accept();

                        connectionsAccepted++;

                        if (sc != null) {

                            sc.configureBlocking(false);

                            synchronized (pkeys) {

                               sc.register(pollSelector, SelectionKey.OP_READ);

                            }

                        } else {

                            throw new RuntimeException(

                                "Socket does not support Channels");

                        }

                    }

                }

            }

            acceptKey.cancel();

            requestThread.join();

            acceptSelector.close();

            pollSelector.close();

        }

    }

}

class RemoteEntity {

    private static Random rnd = new Random();

    int id;

    ByteBuffer data;

    int dataWrittenIndex;

    int totalDataLength;

    boolean initiated = false;

    boolean connected = false;

    boolean written = false;

    boolean acked = false;

    boolean closed = false;

    private SocketChannel sc;

    ByteBuffer ackBuffer;

    PrintStream log;

    InetSocketAddress server;

    RemoteEntity(int id, InetSocketAddress server, PrintStream log)

        throws Exception

    {

        int connectFailures = 0;

        this.id = id;

        this.log = log;

        this.server = server;

        sc = SocketChannel.open();

        sc.configureBlocking(false);

        // Prepare the data buffer to write out from this entity

        // Let's use both slow and fast buffers

        if (rnd.nextBoolean())

            data = ByteBuffer.allocateDirect(100);

        else

            data = ByteBuffer.allocate(100);

        String number = Integer.toString(id);

        if (number.length() == 1)

            number = "0"+number;

        String source = "Testing from " + number;

        data.put(source.getBytes("8859_1"));

        data.flip();

        totalDataLength = source.length();

        // Allocate an ack buffer

        ackBuffer = ByteBuffer.allocateDirect(10);

    }

    private void reset() throws Exception {

        sc.close();

        sc = SocketChannel.open();

        sc.configureBlocking(false);

    }

    private void connect() throws Exception {

        try {

            connected = sc.connect(server);

            initiated = true;

        }  catch (ConnectException e) {

            initiated = false;

            reset();

        }

    }

    private void finishConnect() throws Exception {

        try {

            connected = sc.finishConnect();

        }  catch (IOException e) {

            initiated = false;

            reset();

        }

    }

    int id() {

        return id;

    }

    boolean cycle() throws Exception {

        if (!initiated)

            connect();

        else if (!connected)

            finishConnect();

        else if (!written)

            writeCycle();

        else if (!acked)

            ackCycle();

        else if (!closed)

            close();

        return closed;

    }

    private void ackCycle() throws Exception {

        //log.println("acking from "+id);

        int bytesRead = sc.read(ackBuffer);

        if (bytesRead > 0) {

            acked = true;

        }

    }

    private void close() throws Exception {

        sc.close();

        closed = true;

    }

    private void writeCycle() throws Exception {

        log.println("writing from "+id);

        int numBytesToWrite = rnd.nextInt(10)+1;

        int newWriteTarget = dataWrittenIndex + numBytesToWrite;

        if (newWriteTarget > totalDataLength)

            newWriteTarget = totalDataLength;

        data.limit(newWriteTarget);

        int bytesWritten = sc.write(data);

        if (bytesWritten > 0)

            dataWrittenIndex += bytesWritten;

        if (dataWrittenIndex == totalDataLength) {

            written = true;

            sc.socket().shutdownOutput();

        }

    }

}

class RequestHandler implements Runnable {

    private static Random rnd = new Random();

    private Selector selector;

    private int connectionsHandled = 0;

    private HashMap dataBin = new HashMap();

    PrintStream log;

    public RequestHandler(Selector selector, PrintStream log) {

        this.selector = selector;

        this.log = log;

    }

    public void run() {

        log.println("starting request handler...");

        int connectionsAccepted = 0;

        Set nKeys = selector.keys();

        Set readyKeys = selector.selectedKeys();

        try {

            while(connectionsHandled < SelectorTest.NUM_CLIENTS) {

                int numKeys = selector.select(100);

                // Process channels with data

                synchronized (nKeys) {

                    if (readyKeys.size() > 0) {

                        Iterator i = readyKeys.iterator();

                        while(i.hasNext()) {

                            SelectionKey sk = (SelectionKey)i.next();

                            i.remove();

                            SocketChannel sc = (SocketChannel)sk.channel();

                            if (sc.isOpen())

                                read(sk, sc);

                        }

                    }

                }

                // Give other threads a chance to run

                if (numKeys == 0) {

                    try {

                        Thread.sleep(1);

                    } catch (Exception x) {}

                }

            }

        } catch (Exception e) {

            log.println("Unexpected error 1: "+e);

            e.printStackTrace();

        }

    }

    private void read(SelectionKey sk, SocketChannel sc) throws Exception {

        ByteBuffer bin = (ByteBuffer)dataBin.get(sc);

        if (bin == null) {

            if (rnd.nextBoolean())

                bin = ByteBuffer.allocateDirect(100);

            else

                bin = ByteBuffer.allocate(100);

            dataBin.put(sc, bin);

        }

        int bytesRead = 0;

        do {

            bytesRead = sc.read(bin);

        } while(bytesRead > 0);

        if (bytesRead == -1) {

            sk.interestOps(0);

            bin.flip();

            int size = bin.limit();

            byte[] data = new byte[size];

            for(int j=0; j<size; j++)

                data[j] = bin.get();

            String message = new String(data, "8859_1");

            connectionsHandled++;

            acknowledge(sc);

            log.println("Received >>>"+message + "<<<");

            log.println("Handled: "+connectionsHandled);

        }

    }

    private void acknowledge(SocketChannel sc) throws Exception {

            ByteBuffer ackBuffer = ByteBuffer.allocateDirect(10);

            String s = "ack";

            ackBuffer.put(s.getBytes("8859_1"));

            ackBuffer.flip();

            int bytesWritten = 0;

            while(bytesWritten == 0) {

                bytesWritten += sc.write(ackBuffer);

            }

            sc.close();

    }

}