relayclient.java

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

                nextConnectAttemptAt = TimeUtils.toAbsoluteTimeMillis(30 * TimeUtils.ASECOND);
                
                // Get seeds if we need them or the ones we have are old.
                if ((TimeUtils.toRelativeTimeMillis(TimeUtils.timeNow(), gotLastSeedsAt) > (5 * TimeUtils.AMINUTE))
                        || allSeeds.isEmpty()) {
                    allSeeds = new ArrayList<RouteAdvertisement>(Arrays.asList(seedingManager.getActiveSeedRoutes()));
                    gotLastSeedsAt = TimeUtils.timeNow();
                }
                
                // Try seeds until we get a connection, a referral or are closed.
                while ((null == referral) && !allSeeds.isEmpty() && !closed) {
                    RouteAdvertisement aSeed = allSeeds.remove(0);
                    
                    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.
                        Vector<String> seed_eas = aSeed.getDest().getVectorEndpointAddresses();
                        
                        if (!seed_eas.isEmpty()) {
                            EndpointAddress aSeedHost = new EndpointAddress(seed_eas.get(0));
                            
                            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                                LOG.fine("Attempting relay connect to : " + aSeedHost);
                            }
                            
                            referral = connectToRelay(new RelayServerConnection(this, aSeedHost));
                        }
                    } else {
                        // We have a full route, send it to the virtual address of the route!
                        if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                            LOG.fine("Attempting relay connect to : " + aSeed.getDestPeerID());
                        }
                        
                        referral = connectToRelay(new RelayServerConnection(this, aSeed));
                    }
                }
            }
        } catch (Throwable all) {
            if (Logging.SHOW_SEVERE && LOG.isLoggable(Level.SEVERE)) {
                LOG.log(Level.SEVERE, "Uncaught Throwable in thread :" + Thread.currentThread().getName(), all);
            }
        } finally {
            thread = null;
            
            if (Logging.SHOW_INFO && LOG.isLoggable(Level.INFO)) {
                LOG.info("stop client thread");
            }
        }
    }
    
    protected boolean isRelayConnectDone() {
        return (thread == null || Thread.currentThread() != thread);
    }
    
    /**
     *  @param  server  The relay server to connect to
     *  @return The advertisement of an alternate relay server to try.
     */
    RdvAdvertisement connectToRelay(RelayServerConnection server) {
        if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
            LOG.fine("Connecting to " + server);
        }
        
        RdvAdvertisement referral = null;
        
        // make this the current server
        currentServer = server;
        
        // try getting a messenger to the relay peer
        if (!server.createMessenger(leaseLengthToRequest)) {
            return referral;
        }
        
        if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
            LOG.fine("got messenger " + server);
        }
        
        // check the peerId of the relay peer
        if (server.logicalAddress != null && "jxta".equals(server.logicalAddress.getProtocolName())) {
            server.peerId = server.logicalAddress.getProtocolAddress();
        }
        
        // make sure that the peerId was found.
        if (server.peerId == null) {
            if (server.messenger != null) {
                server.sendDisconnectMessage();
                server.messenger.close();
            }
            return referral;
        }
        
        if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
            LOG.fine("got peerId " + server);
        }
        
        synchronized (this) {
            // wait for a response from the server
            // There is no real damage other than bandwidth usage in sending
            // a message on top of the connection request, so we realy do not
            // wait very long before doing it.
            long requestTimeoutAt = TimeUtils.toAbsoluteTimeMillis(5 * TimeUtils.ASECOND);
            
            while (currentServer != null && currentServer.leaseLength == 0 && !isRelayConnectDone()) {
                long waitTimeout = requestTimeoutAt - System.currentTimeMillis();
                
                if (waitTimeout <= 0) {
                    // did not receive the response in time ?
                    break;
                }
                
                try {
                    wait(waitTimeout);
                } catch (InterruptedException e) {
                    // ignore interrupt
                    if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                        LOG.log(Level.FINE, "wait got interrupted early ", e);
                    }
                }
                
                if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                    LOG.fine("wait done");
                }
            }
        }
        
        if (currentServer == null) {
            return server.alternateRelayAdv;
        }
        
        if (isRelayConnectDone()) {
            if (currentServer.messenger != null) {
                currentServer.messenger.close();
            }
            currentServer = null;
            return server.alternateRelayAdv;
        }
        
        // If we did not get a lease in the first 5 secs, maybe it is because
        // the server knows us from a previous session. Then it will wait for
        // a lease renewal message before responding, not just the connection.
        // Send one and wait another 15.
        if (currentServer.leaseLength == 0) {
            
            currentServer.sendConnectMessage(leaseLengthToRequest);
            
            synchronized (this) {
                
                // wait for a response from the server
                long requestTimeoutAt = TimeUtils.toAbsoluteTimeMillis(15 * TimeUtils.ASECOND);
                
                while (currentServer != null && currentServer.leaseLength == 0 && !isRelayConnectDone()) {
                    long waitTimeout = requestTimeoutAt - System.currentTimeMillis();
                    
                    if (waitTimeout <= 0) {
                        // did not receive the response in time ?
                        break;
                    }
                    
                    try {
                        wait(waitTimeout);
                    } catch (InterruptedException e) {
                        // ignore interrupt
                        if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                            LOG.log(Level.FINE, "wait got interrupted early ", e);
                        }
                    }
                    
                    if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                        LOG.fine("wait done");
                    }
                }
            }
        }
        
        // If we had a messenger but are going to give up that relay server because it is
        // not responsive or rejected us. Make sure that the messenger is closed.
        if (currentServer == null) {
            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.fine("did not get connect from " + server);
            }
            // return any alternate relay advertisements
            return server.alternateRelayAdv;
        }
        
        if (currentServer.relayAdv == null || currentServer.leaseLength == 0 || isRelayConnectDone()) {
            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.fine("did not get connect from " + server);
            }
            if (currentServer.messenger != null) {
                currentServer.sendDisconnectMessage();
                currentServer.messenger.close();
            }
            currentServer = null;
            
            // return any alternate relay advertisements
            return server.alternateRelayAdv;
        }
        
        if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
            LOG.fine("Connected to " + server);
        }
        
        RouteAdvertisement holdAdv = server.relayAdv;
        EndpointAddress holdDest = server.logicalAddress;
        
        // register this relay server
        addActiveRelay(holdDest, holdAdv);
        
        // maintain the relay server connection
        referral = maintainRelayConnection(server);
        
        // unregister this relay server
        removeActiveRelay(holdDest, holdAdv);
        
        return referral;
    }
    
    // FIXME: [e-mail] 20030212. This is junk code: that should be a
    // method of RelayServerConnection and at least not refer to currentServer
    // other than to assign the reference.
    protected RdvAdvertisement maintainRelayConnection(RelayServerConnection server) {
        if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
            LOG.fine("maintainRelayConnection() start " + currentServer);
        }
        
        if (server == null) {
            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.fine("RelayConnection() failed at start " + currentServer);
            }
            return null;
        }
        
        synchronized (this) {
            long currentTime = System.currentTimeMillis();
            long renewLeaseAt = currentServer.leaseObtainedAt + currentServer.leaseLength / 3;
            long waitTimeout = 0;
            
            // This will be true if we need to do the first lease renewal early
            // (that is at the time of the next connection check).
            // We'll do that if we did not know the relay server's adv (seed).
            // In that case we told the relay server to send us its own
            // adv, else we told it to send us some alternate adv (we have to
            // chose). In the former case, we want to do a lease connect
            // request soon so that the server has an opportunity to send us
            // the alternate adv that we did not get during initial connection.
            
            boolean earlyRenew = currentServer.seeded;
            
            while (currentServer != null && !isRelayConnectDone()) {
                // calculate how long to wait
                waitTimeout = renewLeaseAt - currentTime;
                
                // check that the waitTimeout is not greater than the messengerPollInterval
                // We want to make sure that we poll. Most of the time it cost nothing.
                // Also, if we urgently need to renew our lease we may wait
                // less, but if we fail to get our lease renewed in time, the
                // delay may become negative. In that case we do not want
                // to start spinning madly. The only thing we can do is just
                // wait some arbitrary length of time for the lease to be
                // renewed. (If that gets badly overdue, we should probably
                // give up on that relay server, though).
                if (waitTimeout > messengerPollInterval || waitTimeout <= 0) {
                    waitTimeout = messengerPollInterval;
                }
                
                if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                    LOG.fine("waitTimeout=" + waitTimeout + " server=" + currentServer);
                }
                
                try {
                    wait(waitTimeout);
                } catch (InterruptedException e) {
                    Thread.interrupted();
                }
                
                if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                    LOG.fine("wait done, server=" + currentServer);
                }
                
                // make sure the server did not disconnect while waiting
                if (currentServer == null) {
                    break;
                }
                
                // get the current time
                currentTime = System.currentTimeMillis();
                
                if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                    LOG.fine("check messenger " + currentServer);
                }
                
                // check if the messenger is still open
                if (currentServer.messenger.isClosed()) {
                    
                    if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                        LOG.fine("Server connection broken");
                    }
                    
                    // See if we can re-open, that happens often.
                    // That's a reason to renew the connection,
                    // Not a reason to give up on the server yet.
                    // Note we do not renew the lease. This is a transient
                    // and if the server forgot about us, it will respond
                    // to the connection alone. Otherwise, we'd rather avoid
                    // getting a response, since in some cases http connections
                    // close after each received message.
                    if (!currentServer.createMessenger(currentServer.leaseLength)) {
                        if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                            LOG.fine("Server connection NOT re-established");
                        }
                        // lost connection to relay server
                        currentServer = null;
                        break;
                    }
                    
                    if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                        LOG.fine("Server connection re-established");
                    }
                    
                    // getMessenger asks for a new lease.
                    // In the meantime, we'll just assume our old lease is
                    // still current and that the messenger breakage was just
                    // a transient.
                    if (!isRelayConnectDone()) {
                        continue;
                    }
                }
                
                // We've been asked to leave. Be nice and tell the
                // server about it.
                if (isRelayConnectDone()) {
                    break;
                }
                
                // check if the lease needs to be renewed
                renewLeaseAt = currentServer.leaseObtainedAt + currentServer.leaseLength / 3;
                if (currentTime >= renewLeaseAt || earlyRenew) {
                    
                    earlyRenew = false;