relayclient.java

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

            
            // make sure the lease is valid
            if (responseLease <= 0) {
                // invalid lease value
                return;
            }
            
            // update the lease values
            currentServer.leaseLength = responseLease;
            currentServer.leaseObtainedAt = System.currentTimeMillis();
            
            // Since we got the lease, if we requested a queue flush, it's
            // now done. We never send it with a new messenger creation, but
            // when the server already has us as a client it does not respond
            // to connections through messenger creation, so we're sure we
            // will have to send an explicit connect message before we get
            // a response. So, we're sure it's done if it was needed.
            currentServer.flushNeeded = false;
            
            if (relayAdv != null) {
                // Set it only if it is the server's own. Else it got
                // published. Still set alternateRelayAdv so that we
                // can return something that could be usefull when this
                // connection breaks.
                PeerID pidOfAdv = relayAdv.getPeerID();
                String pidOfAdvUnique = pidOfAdv.getUniqueValue().toString();

                if (currentServer.peerId.equals(pidOfAdvUnique)) {
                    currentServer.relayAdv = relayAdv;
                    // Fix the embedded route adv !
                    currentServer.relayAdv.getRouteAdv().setDestPeerID(pidOfAdv);
                } else {
                    currentServer.alternateRelayAdv = relayAdv;
                }
            }
            
            notifyAll();
            
        } else if (RelayTransport.DISCONNECTED_RESPONSE.equals(response)) {
            // Disconnect Response
            if (LOG.isEnabledFor(Level.DEBUG)) {
                LOG.debug("disconnected from " + currentServer);
            }
            
            // If our request was denied, the adv that came back is
            // always an alternate one.
            currentServer.alternateRelayAdv = relayAdv;
            
            if (currentServer.messenger != null) {
                currentServer.messenger.close();
            }
            currentServer.messenger = null;
            currentServer.peerId = null;
            currentServer.leaseLength = 0;
            currentServer.leaseObtainedAt = 0;
            currentServer.relayAdv = null;
            currentServer = null;
            notifyAll();
        }
        
        if (LOG.isEnabledFor(Level.DEBUG)) {
            LOG.debug("response handled for " + currentServer);
        }
    }
    
    private static class RelayServerConnection {
        final RelayClient client;
        
        Messenger messenger = null;
        EndpointAddress logicalAddress = null;
        String peerId = null;
        long leaseLength = 0;
        long leaseObtainedAt = 0;
        
        // If seeded out of a raw address, we have relayAddress.
        // relayAdv comes only later.
        RdvAdvertisement relayAdv = null;
        EndpointAddress relayAddress = null;
        
        RdvAdvertisement alternateRelayAdv = null;
        boolean seeded = false;
        boolean flushNeeded = true; // true until we know it's been done
        
        protected RelayServerConnection(RelayClient client, EndpointAddress addr) {
            this.client = client;
            relayAddress = new EndpointAddress(addr, null, null);
            seeded = true;
        }
        
        protected RelayServerConnection(RelayClient client, RdvAdvertisement relayAdv) {
            this.client = client;
            this.relayAdv = relayAdv;
            relayAdv.getRouteAdv().setDestPeerID(relayAdv.getPeerID());
        }
        
        protected boolean createMessenger(long leaseLengthToRequest) {
            
            // make sure the old messenger is closed
            if (messenger != null) {
                messenger.close();
                messenger = null;
            }
            
            List endpointAddresses = null;
            
            // check for a relay advertisement
            if (relayAdv != null) {
                RouteAdvertisement routeAdv = relayAdv.getRouteAdv();

                if (routeAdv != null) {
                    AccessPointAdvertisement accessPointAdv = routeAdv.getDest();

                    if (accessPointAdv != null) {
                        endpointAddresses = accessPointAdv.getVectorEndpointAddresses();
                    }
                }
            } else {
                // silly but if we use getVetorEndpointAddresses, we get
                // strings. It's realy simpler to have only one kind of obj
                // inthere.
                endpointAddresses = new ArrayList(1);
                endpointAddresses.add(relayAddress.toString());
            }
            
            if (LOG.isEnabledFor(Level.DEBUG)) {
                LOG.debug("createMessenger to " + endpointAddresses);
            }
            
            // make sure we found some endpoint addresses to try
            if (endpointAddresses == null) {
                return false;
            }
            
            // try each endpoint address until one is successful
            for (int i = 0; i < endpointAddresses.size(); i++) {
                String s = (String) endpointAddresses.get(i);

                if (s == null) {
                    continue;
                }
                EndpointAddress addr = new EndpointAddress(s);
                
                if (LOG.isEnabledFor(Level.DEBUG)) {
                    LOG.debug("find transport for " + addr);
                }
                
                // get the list of messengers on this endpoint
                Iterator transports = client.endpoint.getAllMessageTransports();
                
                while (transports.hasNext() && messenger == null) {
                    MessageTransport transport = (MessageTransport) transports.next();
                    
                    // only try transports that are senders and allow routing
                    if (transport instanceof MessageSender && ((MessageSender) transport).allowsRouting()) {
                        if (LOG.isEnabledFor(Level.DEBUG)) {
                            LOG.debug("try transport " + transport);
                        }
                        
                        if (addr.getProtocolName().equals(transport.getProtocolName())) {
                            // NOTE: here we're creating a messenger.
                            // For risk management reason, we refrain from
                            // including the flush request at this time in
                            // this. There is the possibility that the
                            // connection will be repeatedly established
                            // by the transport in our bakck, and would keep
                            // including the flush request ! Normaly this
                            // does not matter because the server should
                            // disregard it when it come in that way, but
                            // still, let's be defensive. We will still send
                            // the flush in a subsequent explicit message.
                            String reqStr = RelayTransport.createConnectString(leaseLengthToRequest, (relayAdv == null), false);
                            
                            // NOTE: this is simulating address mangling by CrossgroupMessenger.
                            // The real service param is after the "/" in the below serviceParam arg.
                            EndpointAddress addrToUse = new EndpointAddress(addr, "EndpointService:" + client.groupName,
                                    client.serviceName + "/" + reqStr);

                            messenger = ((MessageSender) transport).getMessenger(addrToUse, null);
                            if (messenger != null && messenger.isClosed()) {
                                messenger = null;
                            }
                            if (messenger != null) {
                                logicalAddress = messenger.getLogicalDestinationAddress();
                                // We're using a known adv, which means that
                                // we did not ask to get the adv back.
                                // Make sure that we do not keep going with
                                // an adv for the wrong peer. That can happen.
                                if (relayAdv != null && !addr2pid(logicalAddress).equals(relayAdv.getPeerID())) {
                                    // oops, wrong guy !
                                    messenger.close();
                                    messenger = null;
                                    logicalAddress = null;
                                }

                                // In case it was not given, set relayAddress
                                // for toString purposes.
                                relayAddress = addr;
                            }
                        }
                    }
                }
            }
            
            if (LOG.isEnabledFor(Level.DEBUG)) {
                LOG.debug("messenger=" + messenger);
            }
            
            return (messenger != null);
        }
        
        protected boolean sendConnectMessage(long leaseLengthToRequest) {
            if (messenger == null || messenger.isClosed()) {
                return false;
            }
            
            Message message = RelayTransport.createConnectMessage(leaseLengthToRequest, (relayAdv == null), flushNeeded);
            
            try {
                messenger.sendMessage(message, "EndpointService:" + client.groupName, client.serviceName + "/" + client.peerId);
            } catch (IOException e) {
                if (LOG.isEnabledFor(Level.WARN)) {
                    LOG.warn("could not send connect message", e);
                }
                
                // connection attempt failed
                return false;
            }
            
            return true;
        }
        
        protected boolean sendDisconnectMessage() {
            if (messenger == null || messenger.isClosed()) {
                return false;
            }
            
            Message message = RelayTransport.createDisconnectMessage();
            
            try {
                messenger.sendMessage(message, "EndpointService:" + client.groupName, client.serviceName + "/" + client.peerId);
            } catch (IOException e) {
                if (LOG.isEnabledFor(Level.WARN)) {
                    LOG.warn("could not send disconnect message", e);
                }
                
                // connection attempt failed
                return false;
            }
            
            return true;
        }
        
        /**
         *  {@inheritDoc}
         **/
        public String toString() {
            
            return
                    ((relayAddress == null) ? "(adv to " + relayAdv.getPeerID() + ")" : relayAddress.toString()) + " [" + leaseLength + ", "
                    + leaseObtainedAt + "] ";
        }
    }
    
    /**
     * Register an active Relay to the endpoint. This is done
     * so the Route Advertisement of the PeerAdvertisement is
     * updated
     **/
    public synchronized boolean addActiveRelayListener(Object service) {
        
        boolean added = false;
        
        if (!activeRelayListeners.contains(service)) {
            if (LOG.isEnabledFor(Level.DEBUG)) {
                LOG.debug("Register group to relay connection " + ((PeerGroup) service).getPeerGroupName());
            }
            
            activeRelayListeners.add(service);
            
            added = true;
        }
        
        return added;
    }
    
    /**
     * Unregister an active Relay to the endpoint. This is done
     * so the Route Advertisement of the PeerAdvertisement is
     * updated
     **/
    public synchronized boolean removeActiveRelayListener(Object service) {
        activeRelayListeners.remove(service);
        
        return true;
    }
    
    /**
     * Notify of a new relay connection
     *
     **/
    public synchronized boolean addActiveRelay(EndpointAddress address, RouteAdvertisement relayRoute) {
        
        if (LOG.isEnabledFor(Level.DEBUG)) {
            LOG.debug("notify add relay connection for " + address);
        }
        
        // need to notify all our listeners
        Iterator e = activeRelayListeners.iterator();

        while (e.hasNext()) {
            PeerGroup pg = (PeerGroup) e.next();