routeresolver.java

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

/*
 * Copyright (c) 2001-2007 Sun Microsystems, Inc.  All rights reserved.
 *  
 *  The Sun Project JXTA(TM) Software License
 *  
 *  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 Sun Microsystems, Inc. for JXTA(TM) technology." 
 *     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.
 *  
 *  JXTA is a registered trademark of Sun Microsystems, Inc. in the United 
 *  States and other countries.
 *  
 *  Please see the license information page at :
 *  <http://www.jxta.org/project/www/license.html> for instructions on use of 
 *  the license in source files.
 *  
 *  ====================================================================
 *  
 *  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 net.jxta.credential.Credential;
import net.jxta.document.*;
import net.jxta.endpoint.EndpointAddress;
import net.jxta.endpoint.OutgoingMessageEvent;
import net.jxta.exception.PeerGroupException;
import net.jxta.id.ID;
import net.jxta.impl.cm.Srdi;
import net.jxta.impl.cm.Srdi.SrdiInterface;
import net.jxta.impl.cm.SrdiIndex;
import net.jxta.impl.protocol.*;
import net.jxta.impl.util.TimeUtils;
import net.jxta.membership.MembershipService;
import net.jxta.peer.PeerID;
import net.jxta.peergroup.PeerGroup;
import net.jxta.platform.Module;
import net.jxta.protocol.*;
import net.jxta.resolver.QueryHandler;
import net.jxta.resolver.ResolverService;
import net.jxta.resolver.SrdiHandler;
import java.util.logging.Level;
import net.jxta.logging.Logging;
import java.util.logging.Logger;

import java.beans.PropertyChangeEvent;
import java.beans.PropertyChangeListener;
import java.io.IOException;
import java.io.Reader;
import java.io.StringReader;
import java.util.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Handles dynamic route resolution.
 */
class RouteResolver implements Module, QueryHandler, SrdiHandler, SrdiInterface {

    /**
     * Logger
     */
    private final static transient Logger LOG = Logger.getLogger(RouteResolver.class.getName());

    /**
     * Router Service Name
     */
    public final static String routerSName = "EndpointRouter";

    private final static String srdiIndexerFileName = "routerSrdi";

    /**
     * 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;

    /**
     * Default dynamic route resolution configuration preference.
     */
    private final static boolean USE_ROUTE_RESOLVER_DEFAULT = true;

    /**
     * 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 = USE_ROUTE_RESOLVER_DEFAULT;

    /**
     * PeerGroup Service Handle
     */
    private PeerGroup group = null;

    /**
     * Resolver service handle
     */
    private ResolverService resolver = null;

    /**
     * membership service
     */
    private MembershipService membership = null;

    /**
     * EndpointRouter pointer
     */
    private EndpointRouter router = null;

    /**
     * local peer ID as a endpointAddress.
     */
    private EndpointAddress localPeerAddr = null;

    /**
     * local Peer ID
     */
    private ID localPeerId = null;

    /**
     * Route CM Persistent cache
     */
    private RouteCM routeCM = null;

    /**
     *  The current resolver query ID. static to make debugging easier.
     */
    private final static AtomicInteger qid = new AtomicInteger(0);

    /**
     * SRDI route index
     */
    private SrdiIndex srdiIndex = null;

    /**
     * SRDI Index
     */
    private Srdi srdi = null;

    /**
     *  Encapsulates current Membership Service credential.
     */
    final static class CurrentCredential {

        /**
         *	The current default credential
         */
        final Credential credential;
        
        /**
         *	The current default credential in serialized XML form.
         */
        final XMLDocument credentialDoc;
        
        CurrentCredential(Credential credential, XMLDocument credentialDoc) {
            this.credential = credential;
            this.credentialDoc = credentialDoc;
        }
    }
    
    /**
     *   The current Membership service default credential.
     */
    CurrentCredential currentCredential;
    
    /**
     *  Listener we use for membership property events.
     */
    private class CredentialListener implements PropertyChangeListener {

        /**
         *  Standard Constructor
         */
        CredentialListener() {}
        
        /**
         *  {@inheritDoc}
         */
        public void propertyChange(PropertyChangeEvent evt) {
            if (MembershipService.DEFAULT_CREDENTIAL_PROPERTY.equals(evt.getPropertyName())) {
                if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                    LOG.fine("New default credential event");
                }

                synchronized (RouteResolver.this) {
                    Credential cred = (Credential) evt.getNewValue();
                    XMLDocument credentialDoc;
                    if (null != cred) {
                        try {
                            credentialDoc = (XMLDocument) cred.getDocument(MimeMediaType.XMLUTF8);
                            currentCredential = new CurrentCredential(cred, credentialDoc);
                        } catch (Exception all) {
                            if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                                LOG.log(Level.WARNING, "Could not generate credential document", all);
                            }
                            currentCredential = null;
                        }
                    } else {
                        currentCredential = null;
                    }
                }
            }
        }
    }

    final CredentialListener membershipCredListener = new CredentialListener();

    /**
     * @param router the router
     */
    RouteResolver(EndpointRouter router) {
        this.router = router;
    }

    /**
     * initialize  routeResolver
     */
    public void init(PeerGroup group, ID assignedID, Advertisement impl) throws PeerGroupException {

        ModuleImplAdvertisement implAdvertisement = (ModuleImplAdvertisement) impl;

        // extract Router service configuration properties
        ConfigParams confAdv = group.getConfigAdvertisement();
        XMLElement paramBlock = null;

        if (confAdv != null) {
            paramBlock = (XMLElement) confAdv.getServiceParam(assignedID);
        }

        if (paramBlock != null) {
            // get our tunable router parameter
            Enumeration param;

            param = paramBlock.getChildren("useRouteResolver");
            if (param.hasMoreElements()) {
                useRouteResolver = Boolean.getBoolean(((XMLElement) param.nextElement()).getTextValue());
            }
        }

        this.group = group;

        localPeerId = group.getPeerID();

        localPeerAddr = EndpointRouter.pid2addr(group.getPeerID());

        if (Logging.SHOW_CONFIG && LOG.isLoggable(Level.CONFIG)) {
            StringBuilder configInfo = new StringBuilder("Configuring Router Transport Resolver : " + assignedID);

            if (implAdvertisement != null) {
                configInfo.append("\n\tImplementation :");
                configInfo.append("\n\t\tModule Spec ID: ").append(implAdvertisement.getModuleSpecID());
                configInfo.append("\n\t\tImpl Description : ").append(implAdvertisement.getDescription());
                configInfo.append("\n\t\tImpl URI : ").append(implAdvertisement.getUri());
                configInfo.append("\n\t\tImpl Code : ").append(implAdvertisement.getCode());
            }

            configInfo.append("\n\tGroup Params :");
            configInfo.append("\n\t\tGroup : ").append(group);
            configInfo.append("\n\t\tPeer ID : ").append(group.getPeerID());

            configInfo.append("\n\tConfiguration:");
            configInfo.append("\n\t\tUse Route Resolver : ").append(useRouteResolver());
            LOG.config(configInfo.toString());
        }
    }

    /**
     * {@inheritDoc}
     */
    public int startApp(String[] arg) {

        resolver = group.getResolverService();

        if (null == resolver) {
            if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                LOG.warning("Endpoint Router start stalled until resolver service available");
            }
            return Module.START_AGAIN_STALLED;
        }

        membership = group.getMembershipService();

        if (null == membership) {
            if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                LOG.warning("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) {
            // register our credential listener.
            membership.addPropertyChangeListener(MembershipService.DEFAULT_CREDENTIAL_PROPERTY, membershipCredListener);
            
            try {
                // set the initial version of the default credential.
                currentCredential = null;
                Credential credential = membership.getDefaultCredential();
                XMLDocument credentialDoc;

                if (null != credential) {
                    credentialDoc = (XMLDocument) credential.getDocument(MimeMediaType.XMLUTF8);
                    currentCredential = new CurrentCredential(credential, credentialDoc);
                }
            } catch (Exception all) {
                if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                    LOG.log(Level.WARNING, "could not get default credential", all);
                }
            }
        }

        // get the RouteCM cache service
        routeCM = router.getRouteCM();