peerview.java

jxta_src_2.41b jxta 2.41b 最新版源码 from www.jxta.org

        if (failure) {
            // This is a failure notification.
            msg.addMessageElement(MESSAGE_NAMESPACE, FAILURE_ELEMENT);
        }
        
        refreshSelf();
        
        RdvAdvertisement radv = content.getRdvAdvertisement();
        
        XMLDocument doc = (XMLDocument) radv.getDocument(MimeMediaType.XMLUTF8);
        String msgName = response ? RESPONSE_ELEMENT_NAME : MESSAGE_ELEMENT_NAME;
        
        MessageElement msge = new TextDocumentMessageElement(msgName, doc, null);
        
        msg.addMessageElement(MESSAGE_NAMESPACE, msge);
        
        if (!content.equals(self)) {
            // This is a cached RdvAdvertisement
            msg.addMessageElement(MESSAGE_NAMESPACE, CACHED_RADV_ELEMENT);
            
            // This message contains an RdvAdvertisement which is not ourself. In that
            // case, it is wise to also send the local route advertisement (as the optional
            // SrcRdvAdv) so the destination might have a better change to access the "content"
            // RendezvousAdv (this peer will then act as a hop).
            
            RouteAdvertisement localra = RendezVousServiceImpl.extractRouteAdv(lastPeerAdv);
            
            if (localra != null) {
                try {
                    XMLDocument radoc = (XMLDocument) localra.getDocument(MimeMediaType.XMLUTF8);
                    
                    msge = new TextDocumentMessageElement(SRCROUTEADV_ELEMENT_NAME, radoc, null);
                    msg.addMessageElement(MESSAGE_NAMESPACE, msge);
                } catch (Exception ez1) {
                    if (LOG.isEnabledFor(Level.WARN)) {
                        LOG.warn("Could not create optional src route adv for " + content, ez1);
                    }
                }
            }
        }
        
        return msg;
    }
    
    /**
     * Invoked by anyone in order to inform the PeerView of a failure
     * of one of the member peers.
     *
     *  @param pid ID of the peer which failed.
     *  @param propagateFailure If <tt>true</tt>then broadcast the failure to 
     *  other peers otherwise only update the local peerview.
     **/
    public void notifyFailure(PeerID pid, boolean propagateFailure) {
        
        PeerViewElement pve = getPeerViewElement(pid);
        
        if (null != pve) {
            notifyFailure(pve, propagateFailure);
        }
    }
    
    /**
     *  Invoked when a peerview member peer becomes unreachable.
     *
     *  @param pve The peer view element
     *  @param propagateFailure If <tt>true</tt>then broadcast the failure to 
     *  other peers otherwise only update the local peerview.
     **/
    void notifyFailure(PeerViewElement pve, boolean propagateFailure) {
        
        if (LOG.isEnabledFor(Level.DEBUG)) {
            LOG.debug("Notifying failure of " + pve);
        }
        
        try {
            boolean removedFromPeerView = removePeerViewElement(pve);
            
            // only propagate if we actually knew of the peer
            propagateFailure &= (removedFromPeerView || (self == pve));
            
            // Notify local listeners
            if (removedFromPeerView) {
                generateEvent(PeerViewEvent.FAIL, pve);
            }
            
            boolean emptyPeerView = localView.isEmpty();
            
            // If the local view has become empty, reset the kicker into
            // a seeding mode.
            if (emptyPeerView && removedFromPeerView) {
                rescheduleKick(true);
            }
            
            if (propagateFailure) {
                // Notify other rendezvous peers that there has been a failure.
                OutputPipe op = localGroupWirePipeOutputPipe;
                
                if (null != op) {
                    if (LOG.isEnabledFor(Level.DEBUG)) {
                        LOG.debug("Propagating failure of " + pve);
                    }
                    
                    send(op, pve, true, true);
                }
            }
        } catch (Exception ez) {
            if (LOG.isEnabledFor(Level.WARN)) {
                LOG.warn("Failure while generating noficiation of failure of PeerView : " + pve, ez);
            }
        }
    }
    
    /**
     * Invoked by the Timer thread to cause each PeerView to initiate
     * a Peer Advertisement exchange.
     */
    private void kick() {
        
        try {
            if (LOG.isEnabledFor(Level.DEBUG)) {
                LOG.debug("Begun kick() in " + group.getPeerGroupID());
            }
            
            // Use seed strategy. (it has its own throttling and resource limiting).
            seed();
            
            // refresh ourself to a peer in our view
            PeerViewElement refreshee = refreshRecipientStrategy.next();
            
            if ((refreshee != null) && (self != refreshee)) {
                if (LOG.isEnabledFor(Level.DEBUG)) {
                    LOG.debug("Refresh " + refreshee);
                }
                send(refreshee, self, false, false);
            }
            
            if (!rdvService.isRendezVous()) {
                return;
            }
            
            // now share an adv from our local view to another peer from our
            // local view.
            
            PeerViewElement recipient = kickRecipientStrategy.next();
            
            if (recipient == null) {
                if (LOG.isEnabledFor(Level.DEBUG)) {
                    LOG.debug("No recipient to send adv ");
                }
                return;
            }
            
            PeerViewElement rpve = kickAdvertisementStrategy.next();
            
            if (rpve == null) {
                if (LOG.isEnabledFor(Level.DEBUG)) {
                    LOG.debug("No adv to send");
                }
                return;
            }
            
            if (rpve.equals(recipient) || self.equals(recipient)) {
                // give up: no point in sending a peer its own adv
                if (LOG.isEnabledFor(Level.DEBUG)) {
                    LOG.debug("adv to send is same as recipient: Nothing to do.");
                }
                return;
            }
            
            if (LOG.isEnabledFor(Level.DEBUG)) {
                LOG.debug("Sending adv " + rpve + " to " + recipient);
            }
            
            send(recipient, rpve, true, false);
        } finally {
            rescheduleKick(false);
        }
    }
    
    /**
     *  Choose a boot level appropriate for the current configuration and state.
     *
     *  @return the new boot level.
     **/
    private int adjustBootLevel() {
        
        boolean areWeHappy = localView.size() >= minHappyPeerView;
        
        // increment boot level faster if we have a reasonable peerview.
        int increment = areWeHappy ? BOOTLEVEL_INCREMENT : BOOTLEVEL_INCREMENT * 2;
        
        int maxbootlevel = rdvService.isRendezVous() ? MAX_RDV_PEER_BOOTLEVEL : MAX_EDGE_PEER_BOOTLEVEL;
        
        // if we don't have a reasonable peerview, we continue to try harder.
        maxbootlevel -= (areWeHappy ? 0 : BOOTLEVEL_INCREMENT);
        
        bootLevel = Math.min(maxbootlevel, bootLevel + increment);
        
        return bootLevel;
    }
    
    private synchronized void rescheduleKick(boolean now) {
        
        if (closed) {
            return;
        }
        
        // Set the next iteration
        try {
            if (now) {
                bootLevel = MIN_BOOTLEVEL;
            } else {
                adjustBootLevel();
            }
            
            long tilNextKick = DEFAULT_BOOTSTRAP_KICK_INTERVAL * ((1L << bootLevel) - 1);
            
            if (LOG.isEnabledFor(Level.DEBUG)) {
                LOG.debug(
                        "Scheduling kick in " + (tilNextKick / TimeUtils.ASECOND) + " seconds at bootLevel " + bootLevel + " in group "
                        + group.getPeerGroupID());
            }
            
            KickerTask task = new KickerTask();
            
            addTask(task, tilNextKick, -1);
        } catch (Exception ez1) {
            if (LOG.isEnabledFor(Level.ERROR)) {
                LOG.error("Cannot set timer. RPV will not work.", ez1);
            }
        }
    }
    
    /**
     * Send our own advertisement to a randomly choosen set of potential
     * rendezvous peers.
     *
     * @param rdvs A list of RdvAdvertisement for the remote peers.
     * @param maxNb is the maximum number of peers to send the advertisement.
     * @return boolean true if there was at least one peer to send the
     * advertisement to. Returns false otherwise.
     **/
    private boolean sendRandomByAdv(List rdvs, int maxNb, long delay) {
        
        if (rdvs.isEmpty()) {
            if (LOG.isEnabledFor(Level.DEBUG)) {
                LOG.debug("No RDV peers to send queries");
            }
            return false;
        }
        
        int counter = Math.min(maxNb, rdvs.size());
        
        while (counter-- > 0) {
            RdvAdvertisement radv = (RdvAdvertisement) rdvs.remove(0);
            PeerViewElement pve = new PeerViewElement(endpoint, radv);
            
            timedSend(self, pve, delay);
        }
        
        return true;
    }
    
    /**
     * Send our own advertisement to a randomly choosen set of potential
     * rendezvous peers.
     *
     * @param rdvs A list of URI or EndpointAddress for the remote peers.
     * @param maxNb is the maximum number of peers to send the advertisement.
     * @return boolean true if there was at least one peer to send the
     * advertisement to. Returns false otherwise.
     **/
    private boolean sendRandomByAddr(List dests, int maxNb, long delay) {
        
        if (dests.isEmpty()) {
            return false;
        }
        
        int counter = Math.min(maxNb, dests.size());
        
        while (counter-- > 0) {
            Object addr = dests.remove(0);
            
            EndpointAddress dest;
            if( addr instanceof URI ) {
                try {
                dest = new EndpointAddress((URI) addr);
                } catch( IllegalArgumentException badAddr ) {
                    if (LOG.isEnabledFor(Level.WARN)) {
                        LOG.warn("bad URI in seed list : " + addr, badAddr );
                    }
                    continue;
                }
            } else {
                dest = (EndpointAddress) addr;
            }
            
            timedSend(self, dest, delay);
        }
        
        return true;
    }
    
    /**
     * Get a List of RdvAdvertisements locally found in Discovery
     * of peers that w ere known to act as Rendezvous.
     *
     * @return List a vector of RdvAdvertisement.
     **/
    
    private List discoverRdvAdverisements() {
        
        List v = new ArrayList();
        
        DiscoveryService discovery = group.getDiscoveryService();
        
        if (discovery == null) {
            if (LOG.isEnabledFor(Level.WARN)) {
                LOG.warn("Discovery is not yet enabled.");
            }
            
            /* This may happen when a group is just joined. */
            return v;
        }
        
        Enumeration rdvs;
        
        try {
            rdvs = discovery.getLocalAdvertisements(DiscoveryService.ADV, RdvAdvertisement.ServiceNameTag, name );
            
            // start an async query for next time around.
            discovery.getRemoteAdvertisements(null, DiscoveryService.ADV, RdvAdvertisement.ServiceNameTag, name, Integer.MAX_VALUE);
        } catch (IOException e) {
            if (LOG.isEnabledFor(Level.WARN)) {
                LOG.warn("Failed getting RdvAdvertisements from local discovery", e);
            }
            return v;
        }
        
        while (rdvs.hasMoreElements()) {
            Advertisement adv = (Advertisement) rdvs.nextElement();
            
            if (adv instanceof RdvAdvertisement) {
                RdvAdvertisement rdv = (RdvAdvertisement) adv;
                
                if (rdv.getGroupID().equals(group.getPeerGroupID()) && !group.ge