destinations.java

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

/*
 * $Id: Destinations.java,v 1.5 2005/09/17 01:01:32 bondolo Exp $
 *
 * Copyright (c) 2004 Sun Microsystems, Inc.  All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 *
 * 1. Redistributions of source code must retain the above copyright
 *    notice, this list of conditions and the following disclaimer.
 *
 * 2. Redistributions in binary form must reproduce the above copyright
 *    notice, this list of conditions and the following disclaimer in
 *    the documentation and/or other materials provided with the
 *    distribution.
 *
 * 3. The end-user documentation included with the redistribution,
 *    if any, must include the following acknowledgment:
 *       "This product includes software developed by the
 *       Sun Microsystems, Inc. for Project JXTA."
 *    Alternately, this acknowledgment may appear in the software itself,
 *    if and wherever such third-party acknowledgments normally appear.
 *
 * 4. The names "Sun", "Sun Microsystems, Inc.", "JXTA" and "Project JXTA"
 *    must not be used to endorse or promote products derived from this
 *    software without prior written permission. For written
 *    permission, please contact Project JXTA at http://www.jxta.org.
 *
 * 5. Products derived from this software may not be called "JXTA",
 *    nor may "JXTA" appear in their name, without prior written
 *    permission of Sun.
 *
 * THIS SOFTWARE IS PROVIDED ``AS IS'' AND ANY EXPRESSED OR IMPLIED
 * WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
 * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
 * DISCLAIMED.  IN NO EVENT SHALL SUN MICROSYSTEMS OR
 * ITS CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
 * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
 * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
 * USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
 * ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
 * OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT
 * OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
 * SUCH DAMAGE.
 *
 * ====================================================================
 *
 * This software consists of voluntary contributions made by many
 * individuals on behalf of Project JXTA.  For more
 * information on Project JXTA, please see
 * <http://www.jxta.org/>.
 *
 * This license is based on the BSD license adopted by the Apache Foundation.
 */
package net.jxta.impl.endpoint.router;


import java.util.ArrayList;
import java.util.List;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.lang.ref.SoftReference;

import net.jxta.endpoint.EndpointAddress;
import net.jxta.endpoint.EndpointService;
import net.jxta.endpoint.Messenger;

import net.jxta.impl.util.TimeUtils;


/**
 * This class is a repository of wisdom regarding destinations. It also provides a messenger if there is one.  Currently, the
 * wisdom is very limited and is only about direct destinations (for which a messenger once existed).  The wisdom that can be
 * obtained is:
 * <ul>
 *
 * <li> is there a messenger at hand (incoming or otherwise).
 *
 * <li> is it likely that one can be made from this end, should the one we have break. (the last attempt succeeded, not only incoming,
 * and that was not long ago).
 *
 * <li> is either of the above true, (are we confident we can get a messenger as of right now one way or the other).
 *
 * <li> are we supposed to send a welcome to that destination (we can't remember having done it).
 *
 * </ul>
 *
 * This could be extended to manage more of the life cycle, such as knowing about messengers
 * being resolved or having failed to. This primitive interface is temporary; it is only meant to replace messengerPool without
 * having to change the router too much.
 **/

class Destinations implements Runnable {

    private Map wisdoms = new HashMap(64);

    private volatile boolean stopped = false;

    private Thread gcThread = null;

    private EndpointService endpoint;

    /**
     * This class stores knowlege about one particular destination.
     * It does not provide any synchronization. This is provided by the Destinations class.
     **/
    class Wisdom {

        /**
         * How long we consider that a past outgoingMessenger is an indication that one is possible in the future.
         **/
        static final long expiration = 10 * TimeUtils.AMINUTE;

        /**
         * The channel we last used, if any. They disappear faster than the canonical, but, as
         * long as the canonical is around, they can be obtained at a near-zero cost.
         **/
        private SoftReference messenger;

        /**
         * The channel we last used if it happens to be an incoming messenger. We keep 
         * a strong reference to it.
         **/
        private Messenger incomingMessenger;

        /**
         * The transport destination address of the messenger we're caching (if not incoming).
         **/
        private EndpointAddress xportDest;

        /**
         * This tells when the outgoing messenger information expires. Incoming messengers have no expiration per se.  We draw no
         * conclusion from their past presence; only current presence. A wisdom is totally expired (and may thus be removed) when
         * its outgoing messenger information is expired AND it has no incoming messenger.
         **/
        private long expiresAt = 0;
        
        /**
         * When a new destination is added, we're supposed to send our welcome along with the first message.
         * This tells whether isWelcomeNeeded was once invoked or not.
         **/
        private boolean welcomeNeeded = true;

        /**
         * @param channel The messenger to cache information about.
         * @param incoming If true, this is an incoming messenger, which means that
         * if the channel is lost it cannot be re-obtained. It must strongly referenced until it
         * closes for disuse, or breaks.
         **/
        Wisdom(Messenger messenger, boolean incoming) {
            if (incoming) {
                addIncomingMessenger(messenger);
            } else {
                addOutgoingMessenger(messenger);
            }
        }

        /**
         * Tells whether a welcome message is needed.
         *
         * The first time we're asked, we say true. Subsequently, always false; assuming it was done when said so (so,
         * ask only if you'll do it).
         * @return true If this is the first time this method is invoked.
         **/
        boolean isWelcomeNeeded() {
            boolean res = welcomeNeeded;

            welcomeNeeded = false;
            return res;
        }

        boolean addIncomingMessenger(Messenger m) {

            // If we have no other incoming, we take it. No questions asked.
            Messenger currentIncoming = getIncoming();

            if (currentIncoming == null) {
                incomingMessenger = m;
                return true;
            }

            // If it is a relay msgr, we take it too (and drop whatever we had).
            String originAddr = m.getDestinationAddress().getProtocolAddress();
            EndpointAddress logDest = m.getLogicalDestinationAddress();   
        
            if (originAddr.equals(logDest.getProtocolAddress())) {
                incomingMessenger = m;
                return true;
            }
                
            // Now, check reachability. If the old one looks better, prefer it.

            // Compute reachability of the new one.
            int srcPort = Integer.parseInt(originAddr.substring(originAddr.lastIndexOf(':') + 1));
            boolean reachable = (srcPort != 0);

            // Compute reachability of the old one.
            originAddr = currentIncoming.getDestinationAddress().getProtocolAddress();
            logDest = currentIncoming.getLogicalDestinationAddress();
            if (originAddr.equals(logDest.getProtocolAddress())) {
                // The old one is a relay messenger or the like. (and not the new one, we checked).
                return false;
            }
            srcPort = Integer.parseInt(originAddr.substring(originAddr.lastIndexOf(':') + 1));
            boolean currentReachable = (srcPort != 0);

            // The new one is less reachable than the old one. Keep the old one.
            if (currentReachable && !reachable) {
                return false;
            }

            incomingMessenger = m;
            return true;
        }

        boolean addOutgoingMessenger(Messenger m) {
            if (getOutgoing() != null) {
                return false;
            }
            this.messenger = new SoftReference(m);
            xportDest = m.getDestinationAddress();
            expiresAt = TimeUtils.toAbsoluteTimeMillis(expiration);
            return true;
        }

        void noOutgoingMessenger() {
            messenger = null;
            xportDest = null;
            expiresAt = 0;
        }

        /**
         * Returns an incoming messenger is there is one that works. Nulls reference to any broken one
         **/
        private Messenger getIncoming() {
            if (incomingMessenger != null) {
                if ((incomingMessenger.getState() & Messenger.USABLE) != 0) {
                    return incomingMessenger;
                }
                incomingMessenger = null;
            }
            return null;
        }

        /**
         * Returns an outgoingMessenger if there is one or one can be made without delay.
         * Renews a broken one if it can be. Refreshes expiration time if a messenger is returned.
         **/
        private Messenger getOutgoing() {

            if (messenger == null) {
                return null;
            }

            // (If messenger is not null, it means that we also have a xportDest).

            Messenger m = (Messenger) messenger.get();

            // If it is gone or broken, try and get a new one.
            if ((m == null) || ((m.getState() & Messenger.USABLE) == 0)) {