incomingunicastserver.java

JXTA&#8482 is a set of open, generalized peer-to-peer (P2P) protocols that allow any networked devi

    }

    /**
     * Daemon where we wait for incoming connections.
     */
    public void run() {
        try {
            if (Logging.SHOW_INFO && LOG.isLoggable(Level.INFO)) {
                LOG.info("Server is ready to accept connections");
            }

            while (!closed) {
                try {
                    if ((null == serverSocChannel) || !serverSocChannel.isOpen()) {
                        openServerSocket();
                        if (null == serverSocChannel) {
                            if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                                LOG.warning("Failed to open Server Channel");
                            }
                            break;
                        }
                    }
                    acceptSelector.select();
                    Iterator<SelectionKey> it = acceptSelector.selectedKeys().iterator();

                    while (it.hasNext()) {
                        SelectionKey key = it.next();

                        // remove it
                        it.remove();
                        if (key.isAcceptable()) {
                            ServerSocketChannel nextReady = (ServerSocketChannel) key.channel();
                            SocketChannel inputSocket = nextReady.accept();

                            if ((inputSocket == null) || (inputSocket.socket() == null)) {
                                continue;
                            }

                            MessengerBuilder builder = new MessengerBuilder(inputSocket, transport);

                            try {
                                transport.executor.execute(builder);
                                transport.incrementConnectionsAccepted();
                            } catch (RejectedExecutionException re) {
                                if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                                    LOG.log(Level.FINE, MessageFormat.format("Executor rejected task : {0}", builder.toString()), re);
                                }
                            }
                        }
                    }
                } catch (ClosedSelectorException cse) {
                    if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                        LOG.fine("Accept Selector closed");
                    }
                    if (closed) {
                        break;
                    }
                } catch (InterruptedIOException woken) {
                    if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                        LOG.fine("Accept Thread woken");
                    }
                } catch (IOException e1) {
                    if (closed) {
                        break;
                    }
                    if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                        LOG.log(Level.WARNING,
                                "[1] ServerSocket.accept() failed on " + serverSocket.getInetAddress() + ":"+ serverSocket.getLocalPort(), e1);
                    }
                } catch (SecurityException e2) {
                    if (closed) {
                        break;
                    }
                    if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                        LOG.log(Level.WARNING, "[2] ServerSocket.accept() failed on " + serverSocket.getInetAddress() + ":" + serverSocket.getLocalPort(), e2);
                    }
                }
            }
        } catch (Throwable all) {
            if (Logging.SHOW_SEVERE && LOG.isLoggable(Level.SEVERE)) {
                LOG.log(Level.SEVERE, "Uncaught Throwable in thread :" + Thread.currentThread().getName(), all);
            }
        } finally {
            synchronized (this) {
                closed = true;
                ServerSocket temp = serverSocket;

                serverSocket = null;
                if (null != temp) {
                    try {
                        temp.close();
                    } catch (IOException ignored) {
                        if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                            LOG.log(Level.FINE, "Exception occurred while closing server socket", ignored);
                        }
                    }
                }
                acceptThread = null;
            }
            if (Logging.SHOW_INFO && LOG.isLoggable(Level.INFO)) {
                LOG.info("Server has been shut down.");
            }
        }
    }

    public List<Integer> rangeCheckShuffle(int start, int end) {
        if ((start < 1) || (start > 65535)) {
            throw new IllegalArgumentException("Invalid start port");
        }

        if ((end < 1) || (end > 65535) || (end < start)) {
            throw new IllegalArgumentException("Invalid end port");
        }

        // fill the inRange array.
        List<Integer> inRange = new ArrayList<Integer>();

        for (int eachInRange = start; eachInRange < end; eachInRange++) {
            inRange.add(eachInRange);
        }
        Collections.shuffle(inRange);
        return inRange;
    }

    private synchronized void openServerSocket() throws IOException, SecurityException {
        serverSocket = null;
        while (true) {
            try {
                synchronized (this) {
                    acceptSelector = SelectorProvider.provider().openSelector();
                    serverSocChannel = ServerSocketChannel.open();
                    InetSocketAddress bindAddress;

                    if (-1 != serverBindPreferedLocalPort) {
                        bindAddress = new InetSocketAddress(serverBindLocalInterface, serverBindPreferedLocalPort);
                        serverSocket = serverSocChannel.socket();
                        int useBufferSize = Math.max(TcpTransport.RecvBufferSize, serverSocket.getReceiveBufferSize());

                        serverSocket.setReceiveBufferSize(useBufferSize);
                        serverSocket.bind(bindAddress, TcpTransport.MaxAcceptCnxBacklog);
                    } else {
                        List<Integer> rangeList = rangeCheckShuffle(serverBindStartLocalPort, serverBindEndLocalPort);

                        while (!rangeList.isEmpty()) {
                            int tryPort = rangeList.remove(0);

                            if (tryPort > serverBindEndLocalPort) {
                                continue;
                            }

                            try {
                                bindAddress = new InetSocketAddress(serverBindLocalInterface, tryPort);
                                serverSocket = serverSocChannel.socket();
                                int useBufferSize = Math.max(TcpTransport.RecvBufferSize, serverSocket.getReceiveBufferSize());

                                serverSocket.setReceiveBufferSize(useBufferSize);
                                serverSocket.bind(bindAddress, TcpTransport.MaxAcceptCnxBacklog);

                                if (Logging.SHOW_INFO && LOG.isLoggable(Level.INFO)) {
                                    LOG.info("ServerSocketChannel bound to " + bindAddress + ":" + tryPort);
                                }
                            } catch (SocketException failed) {
                                // this one is busy. try another.
                            } catch (Error err) {
                                // this can occur on some platforms where 2 instances are listenting on the same port
                            }
                        }
                    }
                }
                try {
                    // set the new channel non-blocking
                    serverSocChannel.configureBlocking(false);
                    serverSocChannel.register(acceptSelector, SelectionKey.OP_ACCEPT);
                } catch (ClosedChannelException cce) {
                    if (Logging.SHOW_FINER && LOG.isLoggable(Level.FINER)) {
                        LOG.log(Level.FINER, "Channel closed.", cce);
                    }
                }
                if (Logging.SHOW_INFO && LOG.isLoggable(Level.INFO)) {
                    LOG.info("Server will accept connections at " + serverSocket.getLocalSocketAddress());
                }
                return;
            } catch (BindException e0) {
                if (-1 != serverBindStartLocalPort) {
                    serverBindPreferedLocalPort = (0 == serverBindStartLocalPort) ? 0 : -1;
                    continue;
                }
                closed = true;
                if (Logging.SHOW_SEVERE && LOG.isLoggable(Level.SEVERE)) {
                    LOG.log(Level.SEVERE, "Cannot bind ServerSocket on " + serverBindLocalInterface + ":" + serverBindPreferedLocalPort, e0);
                }
                return;
            }
        }
    }

    /**
     * An Executor task that creates a messenger from an incoming SocketChannel
     * object.
     */
    private static class MessengerBuilder implements Runnable {

        private SocketChannel inputSocket;
        private TcpTransport transport;
        TcpMessenger newMessenger;

        MessengerBuilder(SocketChannel inputSocket, TcpTransport transport) {
            assert inputSocket.socket() != null;
            this.inputSocket = inputSocket;
            this.transport = transport;
        }

        /**
         * {@inheritDoc}
         */
        public void run() {
            try {
                if (inputSocket != null && inputSocket.isConnected()) {
                    newMessenger = new TcpMessenger(inputSocket, transport);
                }
            } catch (IOException io) {
                // protect against invalid connections
                if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                    LOG.log(Level.FINE, "Messenger creation failure", io);
                }
            } catch (Throwable all) {
                if (Logging.SHOW_SEVERE && LOG.isLoggable(Level.WARNING)) {
                    LOG.log(Level.SEVERE, "Uncaught Throwable", all);
                }
            }
        }
    }
}