routeresolver.java

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

/*
 *
 * $Id: RouteResolver.java,v 1.18 2006/07/24 01:00:24 hamada Exp $
 *
 * Copyright (c) 2001 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.
 */

/**
 * This class is used to manage a persistent CM cache  of route
 * for the router
 */

package net.jxta.impl.endpoint.router;


import java.beans.PropertyChangeEvent;
import java.beans.PropertyChangeListener;
import java.io.Reader;
import java.io.StringReader;
import java.util.Enumeration;
import java.util.Iterator;
import java.util.List;
import java.util.Vector;

import org.apache.log4j.Level;
import org.apache.log4j.Logger;

import net.jxta.credential.Credential;
import net.jxta.document.Advertisement;
import net.jxta.document.AdvertisementFactory;
import net.jxta.document.Element;
import net.jxta.document.MimeMediaType;
import net.jxta.document.StructuredDocument;
import net.jxta.document.StructuredDocumentFactory;
import net.jxta.document.StructuredTextDocument;
import net.jxta.document.TextElement;
import net.jxta.document.XMLDocument;
import net.jxta.endpoint.EndpointAddress;
import net.jxta.endpoint.OutgoingMessageEvent;
import net.jxta.id.ID;
import net.jxta.membership.MembershipService;
import net.jxta.peer.PeerID;
import net.jxta.peergroup.PeerGroup;
import net.jxta.platform.Module;
import net.jxta.protocol.AccessPointAdvertisement;
import net.jxta.protocol.ConfigParams;
import net.jxta.protocol.ResolverQueryMsg;
import net.jxta.protocol.ResolverResponseMsg;
import net.jxta.protocol.ResolverSrdiMsg;
import net.jxta.protocol.RouteAdvertisement;
import net.jxta.protocol.SrdiMessage;
import net.jxta.resolver.QueryHandler;
import net.jxta.resolver.ResolverService;
import net.jxta.resolver.SrdiHandler;

import net.jxta.exception.PeerGroupException;

import net.jxta.impl.cm.Srdi;
import net.jxta.impl.cm.SrdiIndex;
import net.jxta.impl.cm.Srdi.SrdiInterface;
import net.jxta.impl.protocol.ResolverQuery;
import net.jxta.impl.protocol.ResolverResponse;
import net.jxta.impl.protocol.RouteQuery;
import net.jxta.impl.protocol.RouteResponse;
import net.jxta.impl.protocol.SrdiMessageImpl;
import net.jxta.impl.util.TimeUtils;


class RouteResolver implements QueryHandler, SrdiHandler, SrdiInterface {
    
    /**
     *    Log4j Logger
     */ 
    private static transient final Logger LOG = Logger.getLogger(RouteResolver.class.getName());
    
    /**
     *  Configuration property that disables the usage
     *  of dynamic route resolution. Dynamic routes
     *  will not be discovered. set to true by default
     *  can be overwritten via ConfigParams
     **/
    private boolean useRouteResolver = true;
    
    /**
     * PeerGroup Service Handle
     **/
    private PeerGroup group = null;
    
    /**
     * Resolver service handle
     **/
    private ResolverService resolver = null;
    
    /**
     * EndpointRouter pointer
     */
    private EndpointRouter router = null;
    
    /**
     * local peer ID as a endpointAddress.
     **/
    private EndpointAddress localPeerAddr = null;
    
    /**
     * local Peer ID
     **/
    private ID localPeerId = null;
    
    /**
     * Router Service Name
     **/
    public final static String routerSName = "EndpointRouter";
    
    /**
     * Route CM Persistant cache
     **/
    private RouteCM routeCM = null;
    
    /**
     * Resolver Query unique ID
     **/
    private int qid = 0;
    
    /**
     * membership service
     **/
    private MembershipService membership = null;
    
    /**
     *  Listener we register for credential changes.
     **/
    private CredentialListener membershipCredListener = null;
    
    /**
     *  The credential we will include in queries and responses.
     **/
    private Credential credential = null;
    
    /**
     *  The credential as a document.
     **/
    private StructuredDocument credentialDoc = null;
    
    /**
     * SRDI route index
     *
     **/
    private SrdiIndex srdiIndex = null;
    private final static String srdiIndexerFileName = "routerSrdi";
    
    /**
     * SRDI Index
     **/
    private Srdi srdi = null;
    
    /**
     * Negative Route query acknowledgment
     *
     **/
    private final static int NACKROUTE_QUERYID = -1;
    
    /**
     * Bad route expiration. Amount of time we consider a route bad
     *
     */
    private final static long BADROUTE_EXPIRATION = 2L * TimeUtils.AMINUTE;
    
    /**
     *  Listener we use for membership property events.
     */
    private class CredentialListener implements PropertyChangeListener {
        
        /**
         *  {@inheritDoc}
         **/
        public void propertyChange(PropertyChangeEvent evt) {
            if ("defaultCredential".equals(evt.getPropertyName())) {
                if (LOG.isEnabledFor(Level.DEBUG)) {
                    LOG.debug("New default credential event");
                }
                
                synchronized (RouteResolver.this) {
                    credential = (Credential) evt.getNewValue();
                    credentialDoc = null;
                    
                    if (null != credential) {
                        try {
                            credentialDoc = credential.getDocument(MimeMediaType.XMLUTF8);
                        } catch (Exception all) {
                            if (LOG.isEnabledFor(Level.WARN)) {
                                LOG.warn("Could not generate credential document", all);
                            }
                        }
                    }
                }
            }
        }
    }
    
    /**
     * return routeResolver usage
     */
    protected boolean useRouteResolver() {
        return useRouteResolver;
    }
    
    /**
     * disable routeResolver usage
     */
    protected void disableRouteResolver() {
        useRouteResolver = false;
    }
    
    /**
     * enable routeResolver usage
     */
    protected void enableRouteResolver() {
        useRouteResolver = true;
    }
    
    /**
     * Default constructor
     */
    RouteResolver() {}
    
    /**
     * initialize  routeResolver
     */
    public void init(PeerGroup group, ID assignedID, Advertisement impl, EndpointRouter router)
        throws PeerGroupException {
        
        // extract Router service configuration properties
        ConfigParams confAdv = (ConfigParams) group.getConfigAdvertisement();
        Element paramBlock = null;
        
        if (confAdv != null) {
            paramBlock = confAdv.getServiceParam(assignedID);
        }
        
        if (paramBlock != null) {
            // get our tunable router parameter
            Enumeration param;
            
            param = paramBlock.getChildren("useRouteResolver");
            if (param.hasMoreElements()) {
                useRouteResolver = Boolean.getBoolean(((TextElement) param.nextElement()).getTextValue());
            }
        }
        
        this.group = group;
        this.router = router;
        
        localPeerId = group.getPeerID();
        localPeerAddr = new EndpointAddress("jxta", localPeerId.getUniqueValue().toString(), null, null);
        
    }
    
    /**
     * Make this transport as up and running.
     *
     * <p/>When this method is called, all the services are already registered
     * with the peergroup. So we do not need to delay binding any further.
     * All the public methods, which could be called between init and startApp
     * are defensive regarding the services possibly not being there.
     */
    public int startApp(String[] arg) {
        
        resolver = group.getResolverService();
        membership = group.getMembershipService();
        
        if (null == resolver) {
            if (LOG.isEnabledFor(Level.DEBUG)) {
                LOG.debug("Endpoint Router start stalled until resolver service available");
            }
            return Module.START_AGAIN_STALLED;
        }
        
        if (null == membership) {
            if (LOG.isEnabledFor(Level.DEBUG)) {
                LOG.debug("Endpoint Router start stalled until membership service available");
            }
            return Module.START_AGAIN_STALLED;
        }
        
        resolver.registerHandler(routerSName, this);
        // create and register the srdi service
        srdiIndex = new SrdiIndex(group, srdiIndexerFileName);
        // Srdi is a thread but we are not going to start,
        // since the service is reactive.
        srdi = new Srdi(group, routerSName, this, srdiIndex, 0, 0);
        resolver.registerSrdiHandler(routerSName, this);
        
        synchronized (this) {
        try {
                credential = membership.getDefaultCredential();
            
            if (null != credential) {
                credentialDoc = credential.getDocument(MimeMediaType.XMLUTF8);
                } else {
                    credentialDoc = null;
                }
            } catch (Exception all) {
                if (LOG.isEnabledFor(Level.WARN)) {
                    LOG.warn("could not get credential", all);
            }
            }
            
            membershipCredListener = new CredentialListener();
            membership.addPropertyChangeListener("defaultCredential", membershipCredListener);
        }
        
        // get the RouteCM cache service
        routeCM = router.getRouteCM();