endpointserviceimpl.java

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

/*
 *
 * $Id: EndpointServiceImpl.java,v 1.120 2005/11/21 21:05:51 bondolo 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.
 */

package net.jxta.impl.endpoint;


import java.lang.ref.Reference;
import java.lang.ref.SoftReference;

import java.util.Arrays;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Enumeration;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.WeakHashMap;
import java.util.Vector;

import java.io.IOException;

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

import net.jxta.discovery.DiscoveryService;
import net.jxta.document.Advertisement;
import net.jxta.document.AdvertisementFactory;
import net.jxta.document.MimeMediaType;
import net.jxta.document.StructuredDocument;
import net.jxta.document.StructuredDocumentFactory;
import net.jxta.document.StructuredDocumentUtils;
import net.jxta.document.XMLDocument;
import net.jxta.document.XMLElement;
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.MessageFilterListener;
import net.jxta.endpoint.MessageReceiver;
import net.jxta.endpoint.MessageSender;
import net.jxta.endpoint.MessageTransport;
import net.jxta.endpoint.Messenger;
import net.jxta.endpoint.ThreadedMessenger;
import net.jxta.endpoint.ChannelMessenger;
import net.jxta.endpoint.MessengerEvent;
import net.jxta.endpoint.MessengerEventListener;
import net.jxta.endpoint.StringMessageElement;
import net.jxta.id.ID;
import net.jxta.meter.MonitorResources;
import net.jxta.peergroup.PeerGroup;
import net.jxta.peergroup.PeerGroupID;
import net.jxta.protocol.AccessPointAdvertisement;
import net.jxta.protocol.ConfigParams;
import net.jxta.protocol.ModuleImplAdvertisement;
import net.jxta.protocol.PeerAdvertisement;
import net.jxta.protocol.RouteAdvertisement;
import net.jxta.service.Service;

import net.jxta.exception.PeerGroupException;

import net.jxta.impl.meter.MonitorManager;
import net.jxta.impl.endpoint.endpointMeter.EndpointMeter;
import net.jxta.impl.endpoint.endpointMeter.EndpointMeterBuildSettings;
import net.jxta.impl.endpoint.endpointMeter.EndpointServiceMonitor;
import net.jxta.impl.endpoint.endpointMeter.InboundMeter;
import net.jxta.impl.endpoint.endpointMeter.OutboundMeter;
import net.jxta.impl.endpoint.endpointMeter.PropagationMeter;
import net.jxta.impl.endpoint.relay.RelayClient;
import net.jxta.impl.endpoint.router.EndpointRouter;
import net.jxta.impl.util.FastHashMap;
import net.jxta.impl.util.SequenceIterator;


/**
 * This class implements the frontend for all the JXTA  endpoint protocols, as
 * well as the API for the implementation of the core protocols that use
 * directly the EndpointService. It theory it only needs to implement core methods.
 * legacy or convenience methods should stay out. However, that would require
 * a two-level interface for the service (internal and public). May be later.
 **/
public class EndpointServiceImpl implements EndpointService, MessengerEventListener {
    
    /**
     *  Log4J Category
     **/
    private static final Logger LOG = Logger.getLogger(EndpointServiceImpl.class.getName());
    
    // // constants ////
    
    /**
     *  The Wire Message Format we will use by default.
     **/
    public static final MimeMediaType DEFAULT_MESSAGE_TYPE = new MimeMediaType("application/x-jxta-msg").intern();
    
    /**
     *  The name of this service.
     **/
    public static final String ENDPOINTSERVICE_NAME = "EndpointService";
    
    /**
     *  The Message empty namespace. This namespace is reserved for use by
     *  applications. It will not be used by core protocols.
     **/
    public static final String  MESSAGE_EMPTY_NS = "";
    
    /**
     *  The Message "jxta" namespace. This namespace is reserved for use by
     *  core protocols. It will not be used by applications.
     **/
    public static final String  MESSAGE_JXTA_NS = "jxta";
    
    /**
     *  Namespace in which the message source address will be placed.
     **/
    public static final String  MESSAGE_SOURCE_NS = MESSAGE_JXTA_NS;
    
    /**
     *  Element name in which the message source address will be placed.
     **/
    public static final String  MESSAGE_SOURCE_NAME = "EndpointSourceAddress";
    
    /**
     *  Namespace in which the message destination address will be placed.
     **/
    public static final String  MESSAGE_DESTINATION_NS = MESSAGE_JXTA_NS;
    
    /**
     *  Element name in which the message destination address will be placed.
     *  This element is used for loopback detection during propagate. Only
     *  propagate messages currently contain this element.
     **/
    public static final String  MESSAGE_DESTINATION_NAME = "EndpointDestinationAddress";
    
    /**
     *  Namespace in which the message source peer address will be placed.
     **/
    public static final String  MESSAGE_SRCPEERHDR_NS = MESSAGE_JXTA_NS;
    
    /**
     *  Element name in which the message source peer address will be placed.
     *  This element is used for loopback detection during propagate. Only
     *  propagated messages currently contain this element.
     **/
    public static final String  MESSAGE_SRCPEERHDR_NAME = "EndpointHeaderSrcPeer";
    
    EndpointServiceMonitor endpointServiceMonitor;
    
    /**
     *  if true then this service has been initialized
     **/
    private boolean initialized = false;
    
    /** the EndpointMeter   **/
    private EndpointMeter endpointMeter;
    private PropagationMeter propagationMeter;
    
    /** tunable: the virtual messenger queue size */
    private int vmQueueSize = 20;
    
    /** tunable: should the parent endpoint be used? */
    private boolean useParentEndpoint = true;
    
    private ModuleImplAdvertisement implAdv = null;
    private ID assignedID = null;
    
    private PeerGroup group = null;
    private String localPeerId = null;
    private EndpointService parentEndpoint = null;
    private String myServiceName = null;
    private PeerGroup parentGroup = null;
    
    /**
     *  The Message Transports which are registered for this endpoint. This is
     *  only the message transport registered locally, it does not include
     *  transports which are used from other groups.
     *  We very rarely add or remove anything and we have our own policy
     *  for duplicates
     *
     *  <p/>Elements are {@see net.jxta.endpoint.MessageTransport}
     **/
    private final ArrayList messageTransports = new ArrayList();
    
    /**
     * Passive listeners for messengers. Three priority sets, so far.
     * Insertion and removal time is not critical. What is critical is
     * getting a copy of the set of elements, and, marginally the iteration
     * through whatever the result is. The best bet seems to use a vector
     * since, all operations we perform on them is synchronized and it
     * is backed by an array. However we do not use the legacy methods.
     **/
    private final List[] passiveMessengerListeners = { new Vector(), new Vector(), new Vector() };
    
    /**
     * The set of listener managed by this instance of the endpoint svc.
     *
     *  <p/>keys are {@see java.lang.String}
     *  <p/>values are {@see net.jxta.endpoint.EndpointListener}
     **/
    private final Map incomingMessageListeners = new FastHashMap(16);
    
    /**
     * The set of shared transport messengers currently ready for use.
     **/
    private final Map messengerMap = new WeakHashMap(32);
    
    /**
     *  The filter listeners.
     *
     *  We rarely add/remove, never remove without iterating
     *  and insert objects that are always unique. So using a set
     *  does not make sense. An array list is the best.
     *  <p/>elements are {@link FilterListenerAndMask}
     **/
    private final List incomingFilterListeners = new ArrayList();
    private final List outgoingFilterListeners = new ArrayList();
    
    /**
     *  Holder for a filter listener and its conditions
     **/
    private static class FilterListenerAndMask {
        public String namespace;
        public String name;
        public MessageFilterListener listener;
        
        public FilterListenerAndMask(MessageFilterListener listener, String namespace, String name) {
            this.namespace = namespace;
            this.name = name;
            this.listener = listener;
        }
        
        public boolean equals(Object target) {
            if (this == target) {
                return true;
            }
            
            if (null == target) {
                return false;
            }
            
            if (target instanceof FilterListenerAndMask) {
                FilterListenerAndMask likeMe = (FilterListenerAndMask) target;
                
                boolean result = (null != namespace) ? (namespace.equals(likeMe.namespace)) : (null == likeMe.namespace);
                
                result &= (null != name) ? (name.equals(likeMe.name)) : (null == likeMe.name);
                result &= (listener == likeMe.listener);
            }
            
            return false;
        }
    }
    
    
    /**
     * A non blocking messenger that obtains a backing (possibly blocking) messenger on-demand.
     **/
    private class CanonicalMessenger extends ThreadedMessenger {
        
        /**
         * If the hint was not used because there already was a transport
         * messenger available, then it is saved here for the next time we
         * are forced to create a new transport messenger by the breakage
         * of the one that's here.
         * The management of hints is a bit inconsistent for now: the hint
         * used may be different dependent upon which invocation created the current canonical messenger
         * and, although we try to use the hint only once (to avoid carrying an invalid hint forever)
         * it may happen that a hint is used long after it was suggested.
         */
        Object hint;
        
        /**
         * The transport messenger that this canonical messenger currently uses.
         **/
        Messenger cachedMessenger = null;
        
        /**
         * Create a new CanonicalMessenger.
         *
         * @param destination who messages should be addressed to
         **/
        public CanonicalMessenger(int vmQueueSize, EndpointAddress destination, EndpointAddress logicalDestination, Object hint, OutboundMeter messengerMeter) {
            super(group.getPeerGroupID(), destination, logicalDestination, vmQueueSize);
            this.hint = hint;
        }