edgepeerrdvservice.java

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

            if (rdvConnection.isConnected()) {
                rdvConnection.setConnected(false);
                sendDisconnect(rdvConnection);
            }
        }
        
        rdvService.generateEvent(requested ? RendezvousEvent.RDVDISCONNECT : RendezvousEvent.RDVFAILED, rdvid);
        
        if (RendezvousMeterBuildSettings.RENDEZVOUS_METERING && (rendezvousServiceMonitor != null)) {
            RendezvousConnectionMeter rendezvousConnectionMeter = rendezvousServiceMonitor.getRendezvousConnectionMeter(
                    (PeerID) rdvid);
            
            rendezvousConnectionMeter.connectionDisconnected();
        }
    }
    
    /**
     *  Send lease request to the specified peer.
     *
     *  @param pConn The peer to which the message should be sent.
     *  @throws IOException Thrown for errors sending the lease request.
     */
    private void sendLeaseRequest(RdvConnection pConn) throws IOException {
        if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
            LOG.fine("Sending Lease request to " + pConn);
        }
        
        RendezvousConnectionMeter rendezvousConnectionMeter = null;
        
        if (RendezvousMeterBuildSettings.RENDEZVOUS_METERING && (rendezvousServiceMonitor != null)) {
            rendezvousConnectionMeter = rendezvousServiceMonitor.getRendezvousConnectionMeter(pConn.getPeerID().toString());
        }
        
        Message msg = new Message();
        
        // The request simply includes the local peer advertisement.
        msg.replaceMessageElement(RendezVousServiceProvider.RDV_MSG_NAMESPACE_NAME
                ,
                new TextDocumentMessageElement(ConnectRequest, getPeerAdvertisementDoc(), null));
        
        pConn.sendMessage(msg, pName, pParam);
        
        if (RendezvousMeterBuildSettings.RENDEZVOUS_METERING && (rendezvousConnectionMeter != null)) {
            rendezvousConnectionMeter.beginConnection();
        }
    }
    
    /**
     * Description of the Method
     *
     * @param msg Description of Parameter
     */
    private void processConnectedReply(Message msg) {
        
        // get the Peer Advertisement of the RDV.
        MessageElement peerElem = msg.getMessageElement(RendezVousServiceProvider.RDV_MSG_NAMESPACE_NAME, ConnectedRdvAdvReply);
        
        if (null == peerElem) {
            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.fine("Missing rendezvous peer advertisement");
            }
            
            return;
        }
        
        long lease;
        
        try {
            MessageElement el = msg.getMessageElement(RendezVousServiceProvider.RDV_MSG_NAMESPACE_NAME, ConnectedLeaseReply);
            
            if (el == null) {
                if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                    LOG.fine("missing lease");
                }
                return;
            }
            lease = Long.parseLong(el.toString());
        } catch (Exception e) {
            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.log(Level.FINE, "Parse lease failed with ", e);
            }
            return;
        }
        
        ID pId;
        MessageElement el = msg.getMessageElement(RendezVousServiceProvider.RDV_MSG_NAMESPACE_NAME, ConnectedPeerReply);
        
        if (el == null) {
            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.fine("missing rdv peer");
            }
            return;
        }
        
        try {
            pId = IDFactory.fromURI(new URI(el.toString()));
        } catch (URISyntaxException badID) {
            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.fine("Bad RDV peer ID");
            }
            return;
        }
        
        if (lease <= 0) {
            removeRdv(pId, false);
        } else {
            if (rendezVous.containsKey(pId) || (rendezVous.size() < MAX_RDV_CONNECTIONS)) {
                PeerAdvertisement padv = null;
                
                try {
                    XMLDocument asDoc = (XMLDocument) StructuredDocumentFactory.newStructuredDocument(peerElem);
                    
                    padv = (PeerAdvertisement) AdvertisementFactory.newAdvertisement(asDoc);
                } catch (Exception failed) {
                    if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                        LOG.warning("Failed processing peer advertisement");
                    }
                }
                
                if (null == padv) {
                    if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                        LOG.fine("Missing rendezvous peer advertisement");
                    }
                    return;
                }
                
                if (!seedingManager.isAcceptablePeer(padv)) {
                    if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                        LOG.fine("Rejecting lease offer from unacceptable peer : " + padv.getPeerID());
                    }
                    
                    // XXX bondolo 20061123 perhaps we should send a disconnect here.
                    
                    return;
                }
                
                addRdv(padv, lease);
                
                try {
                    DiscoveryService discovery = group.getDiscoveryService();
                    
                    if (null != discovery) {
                        // This is not our own peer adv so we choose not to share it and keep it for only a short time.
                        discovery.publish(padv, lease * 2, 0);
                    }
                } catch (IOException e) {
                    if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                        LOG.log(Level.FINE, "failed to publish Rendezvous Advertisement", e);
                    }
                }
                
                String rdvName = padv.getName();
                
                if (null == padv.getName()) {
                    rdvName = pId.toString();
                }
                
                if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                    LOG.fine("RDV Connect Response : peer=" + rdvName + " lease=" + lease + "ms");
                }
            } else {
                if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                    LOG.fine("Ignoring lease offer from " + pId);
                }
                // XXX bondolo 20040423 perhaps we should send a disconnect here.
            }
        }
    }
    
    /**
     * A timer task for monitoring our active rendezvous connections.
     * <p/>
     * Checks leases, initiates lease renewals, starts new lease requests.
     */
    private class MonitorTask extends TimerTask {
        
        /**
         * @inheritDoc
         */
        @Override
        public void run() {
            try {
                if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                    LOG.fine("[" + group + "] Periodic rendezvous check");
                }
                
                if (closed) {
                    return;
                }
                
                if (!PeerGroupID.worldPeerGroupID.equals(group.getPeerGroupID())) {
                    MessageTransport router = rdvService.endpoint.getMessageTransport("jxta");
                    
                    if (null == router) {
                        if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                            LOG.warning("Rendezvous connection stalled until router is started!");
                        }
                        
                        // Reschedule another run very soon.
                        timer.schedule(new MonitorTask(), 2 * TimeUtils.ASECOND);
                        return;
                    }
                }
                
                List<RdvConnection> currentRdvs = new ArrayList<RdvConnection>(rendezVous.values());
                
                for (RdvConnection pConn : currentRdvs) {
                    try {
                        if (!pConn.isConnected()) {
                            if (Logging.SHOW_INFO && LOG.isLoggable(Level.INFO)) {
                                LOG.fine("[" + group.getPeerGroupID() + "] Lease expired. Disconnected from " + pConn);
                            }
                            removeRdv(pConn.getPeerID(), false);
                            continue;
                        }
                        
                        if (TimeUtils.toRelativeTimeMillis(pConn.getRenewal()) <= 0) {
                            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                                LOG.fine("[" + group.getPeerGroupID() + "] Attempting lease renewal for " + pConn);
                            }
                            
                            sendLeaseRequest(pConn);
                        }
                    } catch (Exception e) {
                        if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                            LOG.log(Level.WARNING, "[" + group.getPeerGroupID() + "] Failure while checking " + pConn, e);
                        }
                    }
                }
                
                // Not enough Rdvs? Try finding more.
                if (rendezVous.size() < MAX_RDV_CONNECTIONS) {
                    if (seeds.isEmpty()) {                       
                        seeds.addAll(Arrays.asList(EdgePeerRdvService.this.seedingManager.getActiveSeedRoutes()));
                    }
                    
                    int sentLeaseRequests = 0;
                    
                    while (!seeds.isEmpty() && (sentLeaseRequests < 3)) {
                        RouteAdvertisement aSeed = seeds.remove(0);
                        
                        Message msg = new Message();
                        
                        // The lease request simply includes the local peer advertisement.
                        msg.addMessageElement(RendezVousServiceProvider.RDV_MSG_NAMESPACE_NAME
                                ,
                                new TextDocumentMessageElement(ConnectRequest, getPeerAdvertisementDoc(), null));
                        
                        Messenger msgr = null;
                        
                        if (null == aSeed.getDestPeerID()) {
                            // It is an incomplete route advertisement. We are going to assume that it is only a wrapper for a single ea.
                            List<String> seed_eas = aSeed.getDest().getVectorEndpointAddresses();
                            
                            if (!seed_eas.isEmpty()) {
                                EndpointAddress aSeedHost = new EndpointAddress(seed_eas.get(0));
                                
                                msgr = rdvService.endpoint.getMessengerImmediate(aSeedHost, null);
                            }
                        } else {
                            // We have a full route, send it to the virtual address of the route!
                            EndpointAddress aSeedHost = new EndpointAddress(aSeed.getDestPeerID(), null, null);
                            
                            msgr = rdvService.endpoint.getMessengerImmediate(aSeedHost, aSeed);
                        }
                        
                        if (null != msgr) {
                            try {
                                msgr.sendMessageN(msg, pName, pParam);
                                sentLeaseRequests++;
                            } catch (Exception failed) {
                                // ignored
                                ;
                            }
                        }
                    }
                } else {
                    // We don't need any of the current seeds. Get new ones when we need them.
                    seeds.clear();
                }
            } catch (Throwable t) {
                if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                    LOG.log(Level.WARNING, "Uncaught throwable in thread :" + Thread.currentThread().getName(), t);
                }
            }
        }
    }
}