rendezvousserviceimpl.java

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

    public boolean addPropagateListener(String serviceName, String serviceParam, EndpointListener listener) {
        if (null == endpoint) {
            throw new IllegalStateException("Unable to register propagate listener. (not started)");
        }

        return endpoint.addIncomingMessageListener(listener, serviceName, serviceParam);
    }

    /**
     * {@inheritDoc}
     */
    public EndpointListener removePropagateListener(String serviceName, String serviceParam, EndpointListener listener) {

        if (null == endpoint) {
            throw new IllegalStateException("Unable to remove propagate listener. (not started)");
        }

        EndpointListener removed = endpoint.removeIncomingMessageListener(serviceName, serviceParam);

        if ((removed != listener) && (null != removed)) {
            // Not the listener we expected.
            // It's kind of bad that we removed it at all, but putting it back should fix things.
            endpoint.addIncomingMessageListener(removed, serviceName, serviceParam);
            return null;
        }
        return listener;
    }

    /**
     * {@inheritDoc}
     */
    public void propagate(Message msg, String serviceName, String serviceParam, int defaultTTL) throws IOException {

        RendezVousServiceProvider currentProvider = provider;

        if (null == currentProvider) {
            throw new IOException("No RDV provider");
        }
        currentProvider.propagate(msg, serviceName, serviceParam, defaultTTL);
    }

    /**
     * {@inheritDoc}
     */
    public void propagate(Enumeration<? extends ID> destPeerIDs, Message msg, String serviceName, String serviceParam, int defaultTTL) throws IOException {

        RendezVousServiceProvider currentProvider = provider;

        if (null == currentProvider) {
            throw new IOException("No RDV provider");
        }
        currentProvider.propagate(destPeerIDs, msg, serviceName, serviceParam, defaultTTL);
    }

    /**
     * {@inheritDoc}
     */
    public void walk(Message msg, String serviceName, String serviceParam, int defaultTTL) throws IOException {
        RendezVousServiceProvider currentProvider = provider;

        if (null == currentProvider) {
            throw new IOException("No RDV provider");
        }
        currentProvider.walk(msg, serviceName, serviceParam, defaultTTL);
    }

    /**
     * {@inheritDoc}
     */
    public void walk(Vector<? extends ID> destPeerIDs, Message msg, String serviceName, String serviceParam, int defaultTTL) throws IOException {

        RendezVousServiceProvider currentProvider = provider;

        if (null == currentProvider) {
            throw new IOException("No RDV provider");
        }
        currentProvider.walk(destPeerIDs, msg, serviceName, serviceParam, defaultTTL);
    }

    /**
     * {@inheritDoc}
     */
    public Vector<RdvAdvertisement> getLocalWalkView() {

        Vector<RdvAdvertisement> result = new Vector<RdvAdvertisement>();

        PeerView currView = getPeerView();

        if (null == currView) {
            return result;
        }

        Collection<PeerViewElement> allPVE = new ArrayList<PeerViewElement>(currView.getView());

        for (PeerViewElement pve : allPVE) {
            RdvAdvertisement adv = pve.getRdvAdvertisement();
            result.add(adv);
        }

        return result;
    }

    /**
     * Returns the PeerView
     *
     * @return the PeerView
     */
    public PeerView getPeerView() {
        RendezVousServiceProvider currentProvider = provider;

        if (currentProvider instanceof RdvPeerRdvService) {
            return ((RdvPeerRdvService) currentProvider).rpv;
        } else {
            return null;
        }
    }

    /**
     * {@inheritDoc}
     */
    public void propagateToNeighbors(Message msg, String serviceName, String serviceParam, int ttl) throws IOException {
        RendezVousServiceProvider currentProvider = provider;

        if (null == currentProvider) {
            throw new IOException("No RDV provider");
        }
        currentProvider.propagateToNeighbors(msg, serviceName, serviceParam, ttl);
    }

    /**
     * {@inheritDoc}
     */
    public void propagateInGroup(Message msg, String serviceName, String serviceParam, int ttl) throws IOException {
        RendezVousServiceProvider currentProvider = provider;
        if (null == currentProvider) {
            throw new IOException("No RDV provider");
        }
        currentProvider.propagateInGroup(msg, serviceName, serviceParam, ttl);
    }

    /**
     * {@inheritDoc}
     */
    public final void addListener(RendezvousListener listener) {
        eventListeners.add(listener);
    }

    /**
     * {@inheritDoc}
     */
    public final boolean removeListener(RendezvousListener listener) {
        return eventListeners.remove(listener);
    }

    /**
     * Creates a rendezvous event and sends it to all registered listeners.
     *
     * @param type      event type
     * @param regarding event peer ID
     */
    public final void generateEvent(int type, ID regarding) {

        Iterator eachListener = Arrays.asList(eventListeners.toArray()).iterator();
        RendezvousEvent event = new RendezvousEvent(getInterface(), type, regarding);

        if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
            LOG.fine("Calling listeners for " + event);
        }

        while (eachListener.hasNext()) {
            RendezvousListener aListener = (RendezvousListener) eachListener.next();

            try {
                aListener.rendezvousEvent(event);
            } catch (Throwable ignored) {
                if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                    LOG.log(Level.WARNING, "Uncaught Throwable in listener (" + aListener + ")", ignored);
                }
            }
        }
    }

    private synchronized void startWatchDogTimer() {
        stopWatchDogTimer();

        autoRdvTask = new RdvWatchdogTask();

        // Now that we have an Auto-switch flag we only use the higher timeout
        // if auto-switch is off .
        // Set a watchdog, so the peer will become rendezvous if, after rdv_watchdog_interval it
        // still has not connected to any rendezvous peer.
        timer.schedule(autoRdvTask, rdv_watchdog_interval, rdv_watchdog_interval);
    }

    private synchronized void stopWatchDogTimer() {
        RdvWatchdogTask tw = autoRdvTask;

        if (tw != null) {
            autoRdvTask.cancel();
            autoRdvTask = null;
        }
    }

    /**
     * Edge Peer mode connection watchdog.
     */
    private class RdvWatchdogTask extends TimerTask {

        /**
         * {@inheritDoc}
         */
        @Override
        public synchronized void run() {
            try {
                int connectedPeers = getConnectedPeerIDs().size();

                if (!isRendezVous()) {
                    if (0 == connectedPeers) {
                        // This peer has not been able to connect to any rendezvous peer.
                        // become one.

                        // become a rendezvous peer.
                        startRendezVous();
                    }
                } else {
                    // Perhaps we can demote ourselves back to an edge

                    int peerViewSize = getLocalWalkView().size();

                    boolean isManyElementsInPeerView = (peerViewSize > DEMOTION_MIN_PEERVIEW_COUNT);
                    boolean isFewClients = (connectedPeers < DEMOTION_MIN_CLIENT_COUNT);

                    if (isManyElementsInPeerView) {
                        if (connectedPeers == 0) {
                            // Demote ourselves if there are no clients and
                            // there are more than the minimum rendezvous around
                            stopRendezVous();
                        } else if (isFewClients && (RendezVousServiceImpl.random.nextDouble() < DEMOTION_FACTOR)) {
                            // Randomly Demote ourselves if there are few clients and
                            // there are many rendezvous
                            stopRendezVous();
                        }
                    }
                }
            } catch (Throwable all) {
                if (Logging.SHOW_SEVERE && LOG.isLoggable(Level.SEVERE)) {
                    LOG.log(Level.SEVERE, "Uncaught Throwable in Timer : " + Thread.currentThread().getName(), all);
                }
            }
        }
    }

    public boolean isMsgIdRecorded(UUID id) {

        boolean found;

        synchronized (msgIds) {
            found = msgIds.contains(id);
        }

        if (Logging.SHOW_FINER && LOG.isLoggable(Level.FINER)) {
            LOG.finer(id + " = " + found);
        }

        return found;
    }

    /**
     * Checks if a message id has been recorded.
     *
     * @param id message to record.
     * @return {@code true} If message was added otherwise (duplicate)
     *         {@code false}.
     */
    public boolean addMsgId(UUID id) {

        synchronized (msgIds) {
            if (isMsgIdRecorded(id)) {
                // Already there. Nothing to do
                return false;
            }

            if (msgIds.size() < MAX_MSGIDS) {
                msgIds.add(id);
            } else {
                msgIds.set((messagesReceived % MAX_MSGIDS), id);
            }

            messagesReceived++;
        }

        if (Logging.SHOW_FINER && LOG.isLoggable(Level.FINER)) {
            LOG.finer("Added Message ID : " + id);
        }

        return true;
    }

    public UUID createMsgId() {
        return UUIDFactory.newSeqUUID();
    }

    /**
     * Get the current provider. This is for debugging purposes only.
     *
     * @return the provider
     * @deprecated This is private for debugging and diagnostics only.
     */
    @Deprecated
    net.jxta.impl.rendezvous.RendezVousServiceProvider getRendezvousProvider() {
        return provider;
    }
}