resolverserviceimpl.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.resolver;

import net.jxta.credential.Credential;
import net.jxta.document.Advertisement;
import net.jxta.document.MimeMediaType;
import net.jxta.document.StructuredDocument;
import net.jxta.document.StructuredDocumentFactory;
import net.jxta.document.XMLDocument;
import net.jxta.endpoint.*;
import net.jxta.id.ID;
import net.jxta.id.IDFactory;
import net.jxta.impl.endpoint.router.EndpointRouter;
import net.jxta.impl.endpoint.router.RouteControl;
import net.jxta.impl.meter.MonitorManager;
import net.jxta.impl.protocol.ResolverQuery;
import net.jxta.impl.protocol.ResolverResponse;
import net.jxta.impl.protocol.ResolverSrdiMsgImpl;
import net.jxta.impl.resolver.resolverMeter.QueryHandlerMeter;
import net.jxta.impl.resolver.resolverMeter.ResolverMeter;
import net.jxta.impl.resolver.resolverMeter.ResolverMeterBuildSettings;
import net.jxta.impl.resolver.resolverMeter.ResolverServiceMonitor;
import net.jxta.impl.resolver.resolverMeter.SrdiHandlerMeter;
import net.jxta.logging.Logging;
import net.jxta.membership.MembershipService;
import net.jxta.meter.MonitorResources;
import net.jxta.peer.PeerID;
import net.jxta.peergroup.PeerGroup;
import net.jxta.platform.Module;
import net.jxta.protocol.ModuleImplAdvertisement;
import net.jxta.protocol.ResolverQueryMsg;
import net.jxta.protocol.ResolverResponseMsg;
import net.jxta.protocol.ResolverSrdiMsg;
import net.jxta.protocol.RouteAdvertisement;
import net.jxta.rendezvous.RendezVousService;
import net.jxta.rendezvous.RendezVousStatus;
import net.jxta.resolver.QueryHandler;
import net.jxta.resolver.ResolverService;
import net.jxta.resolver.SrdiHandler;

import java.beans.PropertyChangeEvent;
import java.beans.PropertyChangeListener;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.logging.Level;
import java.util.logging.Logger;
import java.util.zip.GZIPInputStream;
import java.util.zip.GZIPOutputStream;

/**
 * Implements the {@link net.jxta.resolver.ResolverService} using the standard
 * JXTA Endpoint Resolver Protocol (ERP).
 *
 * @see net.jxta.resolver.ResolverService
 * @see <a href="http://spec.jxta.org/v1.0/docbook/JXTAProtocols.html#proto-erp">JXTA Protocols Specification : Endpoint Resolver Protocol</a>
 */
public class ResolverServiceImpl implements ResolverService {

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

    /**
     * Resolver query endpoint postfix
     */
    public final static String outQueNameShort = "ORes";

    /**
     * Resolver response endpoint postfix
     */
    public final static String inQueNameShort = "IRes";

    /**
     * Resolver srdi endpoint postfix
     */
    public final static String srdiQueNameShort = "Srdi";

    /**
     * MIME Type for gzipped SRDI messages.
     */
    private final static MimeMediaType GZIP_MEDIA_TYPE = new MimeMediaType("application/gzip").intern();

    private String outQueName = outQueNameShort;
    private String inQueName = inQueNameShort;
    private String srdiQueName = srdiQueNameShort;

    private String handlerName = null;
    private PeerGroup group = null;
    private ModuleImplAdvertisement implAdvertisement = null;
    private EndpointService endpoint = null;
    private MembershipService membership = null;

    private RouteControl routeControl = null;

    private final Map<String, QueryHandler> handlers = Collections.synchronizedMap(new HashMap<String, QueryHandler>(5));
    private final Map<String, SrdiHandler> srdiHandlers = Collections.synchronizedMap(new HashMap<String, SrdiHandler>(5));

    private EndpointListener queryListener = null;
    private EndpointListener responseListener = null;
    private EndpointListener srdiListener = null;

    private ResolverServiceMonitor resolverServiceMonitor;
    private ResolverMeter resolverMeter;

    /**
     * the resolver interface object
     */
    private ResolverService resolverInterface = 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 (ResolverServiceImpl.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();

    /**
     * Convenience method for constructing an endpoint address from an id
     *
     * @param destPeer peer id
     * @param serv     the service name (if any)
     * @param parm     the service param (if any)
     * @return endpointAddress for this peer id.
     */
    private static EndpointAddress mkAddress(ID destPeer, String serv, String parm) {
        return new EndpointAddress("jxta", destPeer.getUniqueValue().toString(), serv, parm);
    }

    /**
     * {@inheritDoc}
     */
    public void init(PeerGroup group, ID assignedID, Advertisement impl) {
        implAdvertisement = (ModuleImplAdvertisement) impl;

        this.group = group;
        handlerName = assignedID.toString();
        String uniqueStr = group.getPeerGroupID().getUniqueValue().toString();

        outQueName = uniqueStr + outQueNameShort;
        inQueName = uniqueStr + inQueNameShort;
        srdiQueName = uniqueStr + srdiQueNameShort;

        if (ResolverMeterBuildSettings.RESOLVER_METERING) { // Fix-Me: This needs to be moved to startApp() when the load order issue is resolved
            resolverServiceMonitor = (ResolverServiceMonitor) MonitorManager.getServiceMonitor(group,
                    MonitorResources.resolverServiceMonitorClassID);
            if (resolverServiceMonitor != null) {
                resolverMeter = resolverServiceMonitor.getResolverMeter();
            }
        }

        // Tell tell the world about our configuration.
        if (Logging.SHOW_CONFIG && LOG.isLoggable(Level.CONFIG)) {
            StringBuilder configInfo = new StringBuilder("Configuring Resolver Service : " + 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\tIn Queue name: ").append(outQueName);
            configInfo.append("\n\t\tOut Queue name: ").append(inQueName);
            configInfo.append("\n\t\tSRDI Queue name: ").append(srdiQueName);

            LOG.config(configInfo.toString());
        }
    }

    /**
     * {@inheritDoc}
     */
    public int startApp(String[] arg) {
        endpoint = group.getEndpointService();

        if (null == endpoint) {
            if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                LOG.warning("Stalled until there is an endpoint service");
            }
            return Module.START_AGAIN_STALLED;
        }

        membership = group.getMembershipService();

        if (null == membership) {
            if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                LOG.warning("Stalled until there is a membership service");
            }
            return Module.START_AGAIN_STALLED;
        }

        // Register Listeners