edgepeerrdvservice.java

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

     */
    public Enumeration getConnectedPeers() {
        
        return Collections.enumeration(Collections.EMPTY_LIST);
    }
    
    /**
     *  {@inheritDoc}
     */
    public boolean isConnectedToRendezVous() {
        return !rendezVous.isEmpty();
    }
    
    /**
     *  {@inheritDoc}
     */
    public Enumeration getConnectedRendezVous() {
        
        return Collections.enumeration(Arrays.asList(rendezVous.keySet().toArray()));
    }
    
    /**
     *  {@inheritDoc}
     */
    public Enumeration getDisconnectedRendezVous() {
        List result = Arrays.asList(disconnectedRendezVous.toArray());
        
        if (LOG.isEnabledFor(Level.DEBUG)) {
            LOG.debug(result.size() + " rendezvous disconnections.");
        }
        
        return Collections.enumeration(result);
    }
    
    /**
     *  {@inheritDoc}
     */
    public void challengeRendezVous(ID peerid, long delay) {
        
        // If immediate failure is requested, just do it.
        // <code>disconnectFromRendezVous()</code> will at least get the peer
        // removed from the peerView, even if it is not currently a rendezvous
        // of ours. That permits to purge from the peerview rdvs that we try
        // and fail to connect to, faster than the background keep alive done
        // by PeerView itself.
        if (delay <= 0) {
            removeRdv( peerid, false );
            return;
        }
        
        RdvConnection pConn = (RdvConnection) rendezVous.get(peerid);
        
        if (null != pConn) {
            long adjusted_delay = Math.max(0, Math.min(TimeUtils.toRelativeTimeMillis(pConn.getLeaseEnd()), delay));
            
            pConn.setLease(adjusted_delay, adjusted_delay);
        }
    }
    
    /**
     *  {@inheritDoc}
     */
    public void disconnectFromRendezVous(ID peerId) {
        
        removeRdv((PeerID) peerId, false);
    }
    
    /**
     *  {@inheritDoc}
     */
    public void propagate(Message msg, String serviceName, String serviceParam, int ttl) throws IOException {
        
        ttl = Math.min(ttl, MAX_TTL);
        RendezVousPropagateMessage propHdr = updatePropHeader(msg, getPropHeader(msg), serviceName, serviceParam, ttl);
        
        if (null != propHdr) {
            sendToEachConnection(msg, propHdr);
            sendToNetwork(msg, propHdr);
            
            if (RendezvousMeterBuildSettings.RENDEZVOUS_METERING && (rendezvousMeter != null)) {
                rendezvousMeter.propagateToGroup();
            }
        } else {
            if (LOG.isEnabledFor(Level.DEBUG)) {
                LOG.debug("Declining to propagate " + msg + " (No prop header)");
            }
        }
    }
    
    /**
     *  {@inheritDoc}
     */
    public void propagateInGroup(Message msg, String serviceName, String serviceParam, int ttl) throws IOException {
        
        ttl = Math.min(ttl, MAX_TTL);
        RendezVousPropagateMessage propHdr = updatePropHeader(msg, getPropHeader(msg), serviceName, serviceParam, ttl);
        
        if (null != propHdr) {
            sendToEachConnection(msg, propHdr);
            
            if (RendezvousMeterBuildSettings.RENDEZVOUS_METERING && (rendezvousMeter != null)) {
                rendezvousMeter.propagateToGroup();
            }
        } else {
            if (LOG.isEnabledFor(Level.DEBUG)) {
                LOG.debug("Declining to propagate " + msg + " (No prop header)");
            }
        }
    }
    
    /**
     *  {@inheritDoc}
     */
    protected void repropagate(Message msg, RendezVousPropagateMessage propHdr, String serviceName, String serviceParam) {
        
        try {
            propHdr = updatePropHeader(msg, propHdr, serviceName, serviceParam, MAX_TTL);
            
            if (null != propHdr) {
                // Note (hamada): This is an unnecessary operation, and serves 
                // no purpose other than the additional loads it imposes on the
                // rendezvous.  Local subnet network operations should be (and are)
                // sufficient to achieve the goal.
                //sendToEachConnection(msg, propHdr);
                sendToNetwork(msg, propHdr);
            } else {
                if (LOG.isEnabledFor(Level.DEBUG)) {
                    LOG.debug("Declining to repropagate " + msg + " (No prop header)");
                }
            }
        } catch (Exception ez1) {
            // Not much we can do
            if (LOG.isEnabledFor(Level.WARN)) {
                LOG.warn("Could not repropagate " + msg, ez1);
            }
        }
    }
    
    /**
     *  {@inheritDoc}
     */
    public void walk(Message msg, String serviceName, String serviceParam, int ttl) throws IOException {
        
        propagateInGroup(msg, serviceName, serviceParam, ttl);
    }
    
    /**
     *  {@inheritDoc}
     */
    public void walk(Vector destPeerIDs, Message msg, String serviceName, String serviceParam, int ttl) throws IOException {
        
        propagate(destPeerIDs.elements(), msg, serviceName, serviceParam, ttl);
    }
    
    /**
     *  @inheritDoc
     */
    public PeerConnection getPeerConnection(ID peer) {
        return (PeerConnection) rendezVous.get(peer);
    }
    
    /**
     *  @inheritDoc
     */
    protected PeerConnection[] getPeerConnections() {
        return (PeerConnection[]) rendezVous.values().toArray(new PeerConnection[0]);
    }
    
    /**
     *  Attempt to connect to a rendezvous we have not previously connected to.
     *
     *  @param radv Rendezvous advertisement for the Rdv we want to connect to.
     */
    private void newLeaseRequest(RdvAdvertisement radv) throws IOException {
        
        if (LOG.isEnabledFor(Level.DEBUG)) {
            LOG.debug("Sending new lease request to " + radv.getPeerID() );
        }
        
        RendezvousConnectionMeter rendezvousConnectionMeter = null;
        
        if (RendezvousMeterBuildSettings.RENDEZVOUS_METERING && (rendezvousServiceMonitor != null)) {
            rendezvousConnectionMeter = rendezvousServiceMonitor.getRendezvousConnectionMeter( radv.getPeerID().toString() );
        }
        
        EndpointAddress addr = mkAddress(radv.getPeerID(), null, null);
        
        RouteAdvertisement hint = radv.getRouteAdv();
        
        Messenger messenger = rdvService.endpoint.getMessenger(addr, hint);
        
        if (null == messenger) {
            if (LOG.isEnabledFor(Level.WARN)) {
                LOG.warn("Could not get messenger for " + addr );
            }
            
            throw new IOException("Could not connect to " + addr  );
        }
        
        Message msg = new Message();
        
        // The request simply includes the local peer advertisement.
        msg.replaceMessageElement("jxta", new TextDocumentMessageElement(ConnectRequest, getPeerAdvertisementDoc(), null));
        
        messenger.sendMessage(msg, pName, pParam);
        
        if (RendezvousMeterBuildSettings.RENDEZVOUS_METERING && (rendezvousConnectionMeter != null)) {
            rendezvousConnectionMeter.beginConnection();
        }
    }
    
    private void disconnectFromAllRendezVous() {
        Iterator eachRendezvous = Arrays.asList(rendezVous.values().toArray()).iterator();
        
        while (eachRendezvous.hasNext()) {
            try {
                RdvConnection pConn = (RdvConnection) eachRendezvous.next();
                
                disconnectFromRendezVous(pConn.getPeerID());
            } catch (Exception failed) {
                if (LOG.isEnabledFor(Level.WARN)) {
                    LOG.warn("disconnectFromRendezVous failed ", failed);
                }
            }
        }
    }
    
    /**
     *  Handle a disconnection request from a remote peer.
     *
     * @param  msg  Description of Parameter
     */
    private void processDisconnectRequest(Message msg) {
        
        try {
            MessageElement elem = msg.getMessageElement("jxta", DisconnectRequest);
            
            if (null != elem) {
                InputStream is = elem.getStream();
                PeerAdvertisement adv = (PeerAdvertisement) AdvertisementFactory.newAdvertisement(elem.getMimeType(), is);
                
                RdvConnection rdvConnection = (RdvConnection) rendezVous.get(adv.getPeerID());
                
                if (null != rdvConnection) {
                    rdvConnection.setConnected(false);
                    removeRdv(adv.getPeerID(), true);
                } else {
                    if (LOG.isEnabledFor(Level.DEBUG)) {
                        LOG.debug("Ignoring disconnect request from " + adv.getPeerID());
                    }
                }
            }
        } catch (Exception failure) {
            if (LOG.isEnabledFor(Level.WARN)) {
                LOG.warn("Failure processing disconnect request", failure);
            }
        }
    }
    
    /**
     * Add a rendezvous to our collection of rendezvous peers.
     *
     * @param  padv   PeerAdvertisement for the rendezvous peer.
     * @param  lease  The duration of the lease in relative milliseconds.
     */
    private void addRdv(PeerAdvertisement padv, long lease) {
        
        int eventType;
        
        RdvConnection rdvConnection;
        
        synchronized (rendezVous) {
            rdvConnection = (RdvConnection) rendezVous.get(padv.getPeerID());
            
            if (null == rdvConnection) {
                rdvConnection = new RdvConnection(group, rdvService, padv.getPeerID());
                rendezVous.put(padv.getPeerID(), rdvConnection);
                disconnectedRendezVous.remove(padv.getPeerID());
                eventType = RendezvousEvent.RDVCONNECT;
            } else {
                eventType = RendezvousEvent.RDVRECONNECT;
            }
        }
        
        // Check if the peer is already registered.
        if (RendezvousEvent.RDVRECONNECT == eventType) {
            if (LOG.isEnabledFor(Level.INFO)) {
                LOG.info("Renewed RDV lease from " + rdvConnection);
            }
            
            // Already connected, just upgrade the lease
            
            if (RendezvousMeterBuildSettings.RENDEZVOUS_METERING && (rendezvousServiceMonitor != null)) {
                RendezvousConnectionMeter rendezvousConnectionMeter = rendezvousServiceMonitor.getRendezvousConnectionMeter(padv.getPeerID());
                
                rendezvousConnectionMeter.leaseRenewed(lease);
            }
        } else {
            if (LOG.isEnabledFor(Level.INFO)) {
                LOG.info("New RDV lease from " + rdvConnection);
            }
            
            if (RendezvousMeterBuildSettings.RENDEZVOUS_METERING && (rendezvousServiceMonitor != null)) {
                RendezvousConnectionMeter rendezvousConnectionMeter = rendezvousServiceMonitor.getRendezvousConnectionMeter(padv.getPeerID());
                
                rendezvousConnectionMeter.connectionEstablished(lease);
            }
        }
        
        rdvConnection.connect(padv, lease, Math.min(LEASE_MARGIN, (lease / 2)));
        
        rdvService.generateEvent(eventType, padv.getPeerID());
    }
    
    /**
     * Remove the specified rendezvous from our collection of rendezvous.
     *
     * @param  rdvid  the id of the rendezvous to remove.
     */
    private void removeRdv(ID rdvid, boolean requested) {
        
        if (LOG.isEnabledFor(Level.INFO)) {
            LOG.info("Disconnect from RDV " + rdvid );
        }
        
        PeerConnection rdvConnection;
        
        synchronized (this) {
            rdvConnection = (PeerConnection) rendezVous.remove(rdvid);
            
            // let's add it to the list of disconnected rendezvous
            if( null != rdvConnection ) {
                disconnectedRendezVous.add(rdvid);
            }
        }
        
        if (null != rdvConnection) {
            if (rdvConnection.isConnected()) {
                rdvConnection.setConnected(false);
                sendDisconnect( rdvConnection );
            }
        }
        
        /*
         *  Remove the rendezvous we are disconnecting from the peerview as well.
         *  This prevents us from immediately reconnecting to it.
         */
        rdvService.rpv.notifyFailure((PeerID) rdvid, false);
        
        rdvService.generateEvent(requested ? RendezvousEvent.RDVDISCONNECT : RendezvousEvent.RDVFAILED, rdvid);
        
        if (RendezvousMeterBuildSettings.RENDEZVOUS_METERING && (rendezvousServiceMonitor != null)) {
            RendezvousConnectionMeter rendezvousConnectionMeter = rendezvousServiceMonitor.getRendezvousConnectionMeter((PeerID) rdvid);