resolverserviceimpl.java

jxta平台的开发包

/*
 *  Copyright (c) 2001-2006 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.
 *
 * $Id: ResolverServiceImpl.java,v 1.95 2006/10/10 23:41:27 hamada Exp $
 */
package net.jxta.impl.resolver;

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.Iterator;
import java.util.Map;
import java.util.zip.GZIPInputStream;
import java.util.zip.GZIPOutputStream;

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.ByteArrayMessageElement;
import net.jxta.endpoint.EndpointAddress;
import net.jxta.endpoint.EndpointListener;
import net.jxta.endpoint.EndpointService;
import net.jxta.endpoint.Message;
import net.jxta.endpoint.MessageElement;
import net.jxta.endpoint.MessageTransport;
import net.jxta.endpoint.Messenger;
import net.jxta.endpoint.OutgoingMessageEvent;
import net.jxta.endpoint.OutgoingMessageEventListener;
import net.jxta.endpoint.TextDocumentMessageElement;
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.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.resolver.QueryHandler;
import net.jxta.resolver.ResolverService;
import net.jxta.resolver.SrdiHandler;
import net.jxta.service.Service;

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

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

    /**
     *  Log4J 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 myGroup = null;
    private ModuleImplAdvertisement implAdvertisement = null;
    private EndpointService endpoint = null;
    private RendezVousService rendezvous = null;
    private MembershipService membership = null;

    private RouteControl routeControl = null;

    private final Map handlers = Collections.synchronizedMap(new HashMap(5));
    private final Map srdiHandlers = Collections.synchronizedMap(new HashMap(5));

    private CredentialListener membershipCredListener = null;
    private Credential credential = null;
    private StructuredDocument credentialDoc = null;

    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;

    /**
     *  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 (ResolverServiceImpl.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);
                            }
                        }
                    }
                }
            }
        }
    }

    /**
     *  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)
     *  @param endpointAddress for this peer id.
     */
    private final static EndpointAddress mkAddress(ID destPeer, String serv, String parm) {
        EndpointAddress addr = new EndpointAddress("jxta", destPeer.getUniqueValue().toString(), serv, parm);
        return addr;
    }

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

        implAdvertisement = (ModuleImplAdvertisement) impl;

        myGroup = group;
        handlerName = assignedID.toString();
        String uniqueStr = myGroup.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(myGroup, MonitorResources.resolverServiceMonitorClassID);
            if (resolverServiceMonitor != null) {
                resolverMeter = resolverServiceMonitor.getResolverMeter();
            }
        }

        // Tell tell the world about our configuration.
        if (LOG.isEnabledFor(Level.INFO)) {
            StringBuffer configInfo = new StringBuffer("Configuring Resolver Service : " + assignedID);

            configInfo.append("\n\tImplementation:");
            configInfo.append("\n\t\tImpl Description: " + implAdvertisement.getDescription());
            configInfo.append("\n\t\tImpl URI : " + implAdvertisement.getUri());
            configInfo.append("\n\t\tImpl Code : " + implAdvertisement.getCode());

            configInfo.append("\n\tGroup Params:");
            configInfo.append("\n\t\tGroup: " + myGroup.getPeerGroupName());
            configInfo.append("\n\t\tGroup ID: " + myGroup.getPeerGroupID());
            configInfo.append("\n\t\tPeer ID: " + myGroup.getPeerID());

            configInfo.append("\n\tConfiguration:");
            configInfo.append("\n\t\tIn Queue name: " + outQueName);
            configInfo.append("\n\t\tOut Queue name: " + inQueName);
            configInfo.append("\n\t\tSRDI Queue name: " + srdiQueName);

            LOG.info(configInfo);
        }
    }

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

        if (null == endpoint) {
            if (LOG.isEnabledFor(Level.WARN)) {
                LOG.warn("Stalled until there is an endpoint service");
            }

            return Module.START_AGAIN_STALLED;
        }

        membership = myGroup.getMembershipService();

        if (null == membership) {
            if (LOG.isEnabledFor(Level.WARN)) {
                LOG.warn("Stalled until there is a membership service");
            }

            return Module.START_AGAIN_STALLED;
        }

        // FIXME 20040122 bondolo What if Rendezvous just happens to load
        // AFTER resolver? This seems unacceptable.
        rendezvous = myGroup.getRendezVousService();

        // Register Listeners
        try {
            // Register Query Listener
            queryListener = new DemuxQuery();
            if(!endpoint.addIncomingMessageListener(queryListener, handlerName, outQueName)) {
                if (LOG.isEnabledFor(Level.ERROR)) {
                    LOG.error("Cannot register listener (already registered)");
                }
            } else {
                if (null != rendezvous) {
                    if(!rendezvous.addPropagateListener(handlerName + outQueName, queryListener)) {
                        if (LOG.isEnabledFor(Level.ERROR)) {
                            LOG.error("Cannot register query listener (already registered)");
                        }
                    }
                } else {
                    if (LOG.isEnabledFor(Level.ERROR)) {
                        LOG.error("Failed to register query listener (null rendezvous service)");
                    }
                }
            }

            // Register Response Listener
            responseListener = new DemuxResponse();