rdvpeerrdvservice.java

jxta平台的开发包

     **/
    public void disconnectFromRendezVous(ID peerId) {
        
        throw new UnsupportedOperationException("Not supported by rendezvous");
    }
    
    /**
     *  {@inheritDoc}
     **/
    public boolean isConnectedToRendezVous() {
        return false;
    }
    
    /**
     *  {@inheritDoc}
     **/
    public void setChoiceDelay(long delay) {
        // No effect on rendezvous
    }
    
    /**
     *  {@inheritDoc}
     **/
    public Vector getConnectedPeerIDs() {
        
        Vector result = new Vector();
        List allClients = Arrays.asList(clients.values().toArray());
        
        Iterator eachClient = allClients.iterator();
        
        while (eachClient.hasNext()) {
            PeerConnection aConnection = (PeerConnection) eachClient.next();
            
            if (aConnection.isConnected()) {
                result.add(aConnection.getPeerID());
            }
        }
        
        return result;
    }
    
    /**
     *  {@inheritDoc}
     **/
    public Enumeration getConnectedPeers() {
        
        return Collections.enumeration(getConnectedPeerIDs());
    }
    
    /**
     *  {@inheritDoc}
     **/
    public Enumeration getConnectedRendezVous() {
        // This is a rdv peer. Cannot connect to other rdvs.
        return Collections.enumeration(Collections.EMPTY_LIST);
    }
    
    /**
     *  {@inheritDoc}
     **/
    public Enumeration getDisconnectedRendezVous() {
        // This is a rdv peer. Cannot connect to other rdvs.
        return Collections.enumeration(Collections.EMPTY_LIST);
    }
    
    /**
     *  {@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) {
            walk((Message) msg.clone(), PropSName, PropPName, ttl);
            //hamada: this is a very expensive operation and therefore not a supported operation
            //sendToEachConnection(msg, propHdr);
            sendToNetwork(msg, propHdr);
            
            if (RendezvousMeterBuildSettings.RENDEZVOUS_METERING && (rendezvousMeter != null)) {
                rendezvousMeter.propagateToGroup();
            }
        }
    }
    
    /**
     *  {@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) {
            walk((Message) msg.clone(), PropSName, PropPName, ttl);
            sendToEachConnection(msg, propHdr);
            
            if (RendezvousMeterBuildSettings.RENDEZVOUS_METERING && (rendezvousMeter != null)) {
                rendezvousMeter.propagateToGroup();
            }
        }
    }
    
    /**
     *  {@inheritDoc}
     * @deprecated this operation is a privileged operation and expensive 
     * operation to be supported, use walk instead
     **/
    protected void repropagate(Message msg, RendezVousPropagateMessage propHdr, String serviceName, String serviceParam) {
        /*
        try {
            RendezVousPropagateMessage newPropHdr = updatePropHeader(msg, propHdr, serviceName, serviceParam, MAX_TTL);
            
            if (null != newPropHdr) {
                walk((Message) msg.clone(), PropSName, PropPName, MAX_TTL);
                sendToEachConnection(msg, newPropHdr);
                sendToNetwork(msg, newPropHdr);
            } else {
                if (LOG.isEnabledFor(Level.DEBUG)) {
                    LOG.debug("Null propagate header, declining to repropagate " + msg);
                }
            }
        } catch (Exception ez1) {
            // Not much we can do
            if (LOG.isEnabledFor(Level.DEBUG)) {
                LOG.debug("Could not repropagate " + msg + " (" + propHdr.getMsgId() + ")", ez1);
            }
        }
        */
    }
    
    /**
     *  @inheritDoc
     **/
    public PeerConnection getPeerConnection(ID peer) {
        return (PeerConnection) clients.get(peer);
    }
    
    /**
     *  @inheritDoc
     **/
    protected PeerConnection[] getPeerConnections() {
        return (PeerConnection[]) clients.values().toArray(new PeerConnection[0]);
    }
    
    /**
     * Add a client to our collection of clients.
     *
     * @param  padv   The advertisement of the peer to be added.
     * @param  lease  The lease duration in relative milliseconds.
     **/
    private ClientConnection addClient(PeerAdvertisement padv, long lease) {
        ClientConnectionMeter clientConnectionMeter = null;
        
        int eventType;
        ClientConnection pConn;
        
        synchronized (clients) {
            pConn = (ClientConnection) clients.get(padv.getPeerID());
            
            // Check if the peer is already registered.
            if (null != pConn) {
                eventType = RendezvousEvent.CLIENTRECONNECT;
            } else {
                eventType = RendezvousEvent.CLIENTCONNECT;
                pConn = new ClientConnection(group, rdvService, padv.getPeerID());
                clients.put(padv.getPeerID(), pConn);
            }
        }
        
        if (RendezvousMeterBuildSettings.RENDEZVOUS_METERING && (rendezvousServiceMonitor != null)) {
            clientConnectionMeter = rendezvousServiceMonitor.getClientConnectionMeter(padv.getPeerID());
        }
        
        if (RendezvousEvent.CLIENTCONNECT == eventType) {
            if (RendezvousMeterBuildSettings.RENDEZVOUS_METERING && (clientConnectionMeter != null)) {
                clientConnectionMeter.clientConnectionEstablished(lease);
            }
        } else {
            if (RendezvousMeterBuildSettings.RENDEZVOUS_METERING && (clientConnectionMeter != null)) {
                clientConnectionMeter.clientLeaseRenewed(lease);
            }
        }
        
        rdvService.generateEvent(eventType, padv.getPeerID());
        
        pConn.connect(padv, lease);
        
        return pConn;
    }
    
    /**
     *  Removes the specified client from the clients collections.
     *
     *  @param  pConn   The connection object to remove.
     *  @param  requested   If <code>true</code> then the disconnection was
     *  requested by the remote peer.
     *  @return the ClientConnection object of the client or <code>null</code>
     *  if the client was not known.
     */
    private ClientConnection removeClient(PeerConnection pConn, boolean requested) {
        
        if (LOG.isEnabledFor(Level.DEBUG)) {
            LOG.debug("Disconnecting client " + pConn);
        }

        if (pConn.isConnected()) {
            pConn.setConnected(false);
            sendDisconnect( pConn );
        }
        
        rdvService.generateEvent( requested ? RendezvousEvent.CLIENTDISCONNECT : RendezvousEvent.CLIENTFAILED, pConn.getPeerID());      
        
        if (RendezvousMeterBuildSettings.RENDEZVOUS_METERING && (rendezvousServiceMonitor != null)) {
            ClientConnectionMeter clientConnectionMeter = rendezvousServiceMonitor.getClientConnectionMeter((PeerID) pConn.getPeerID());
            
            clientConnectionMeter.clientConnectionDisconnected(requested);
        }
        
        return (ClientConnection) clients.remove(pConn.getPeerID());
    }
    
    private void disconnectAllClients() {
        Iterator eachConnected = Arrays.asList(clients.values().toArray()).iterator();
        
        while (eachConnected.hasNext()) {
            ClientConnection pConn = (ClientConnection) eachConnected.next();
            
            try {
                removeClient(pConn, false);
            } catch (Exception ez1) {
                if (LOG.isEnabledFor(Level.WARN)) {
                    LOG.warn("disconnectClient failed for" + pConn, ez1);
                }
                continue;
            }
        }
    }
    
    /**
     * Handle a disconnection request
     *
     * @param  msg  Message containting the disconnection request.
     */
    private void processDisconnectRequest(Message msg) {
        
        PeerAdvertisement adv = null;
        
        try {
            MessageElement elem = msg.getMessageElement("jxta", DisconnectRequest);
            
            adv = (PeerAdvertisement) AdvertisementFactory.newAdvertisement(elem.getMimeType(), elem.getStream());
        } catch (Exception e) {
            return;
        }
        
        ClientConnection pConn = (ClientConnection) clients.get(adv.getPeerID());
        
        if (null != pConn) {
            removeClient(pConn, true);
        }
    }
    
    /**
     *  Handles a lease request message
     *
     *  @param  msg  Message containing the lease request
     */
    private void processLeaseRequest(Message msg) {
        
        PeerAdvertisement padv;
        
        try {
            MessageElement elem = msg.getMessageElement("jxta", ConnectRequest);
            
            padv = (PeerAdvertisement) AdvertisementFactory.newAdvertisement(elem.getMimeType(), elem.getStream());
            msg.removeMessageElement(elem);
        } catch (Exception e) {
            if (LOG.isEnabledFor(Level.DEBUG)) {
                LOG.debug("Cannot retrieve advertisment from lease request", e);
            }
            return;
        }
        
        // Publish the client's peer advertisement
        try {
            DiscoveryService discovery = group.getDiscoveryService();
            
            if (null != discovery) {
                // This is not our own peer adv so we must not share it or keep it that long.
                discovery.publish(padv, leaseDuration * 2, 0 );
            }
        } catch (Exception e) {
            if (LOG.isEnabledFor(Level.WARN)) {
                LOG.warn("Client peer advertisement publish failed", e);
            }
        }
        
        long lease;
        
        ClientConnection pConn = (ClientConnection) clients.get(padv.getPeerID());
        
        if (null != pConn) {
            if (LOG.isEnabledFor(Level.DEBUG)) {
                LOG.debug("Renewing client lease to " + pConn );
            }
            
            lease = leaseDuration;
        } else {