endpointserviceimpl.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;

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.ChannelMessenger;
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.MessagePropagater;
import net.jxta.endpoint.MessageReceiver;
import net.jxta.endpoint.MessageSender;
import net.jxta.endpoint.MessageTransport;
import net.jxta.endpoint.Messenger;
import net.jxta.endpoint.MessengerEvent;
import net.jxta.endpoint.MessengerEventListener;
import net.jxta.endpoint.StringMessageElement;
import net.jxta.endpoint.ThreadedMessenger;
import net.jxta.exception.PeerGroupException;
import net.jxta.id.ID;
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.endpoint.tcp.TcpTransport;
import net.jxta.impl.meter.MonitorManager;
import net.jxta.impl.util.SequenceIterator;
import net.jxta.logging.Logging;
import net.jxta.meter.MonitorResources;
import net.jxta.peergroup.PeerGroup;
import net.jxta.platform.Module;
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 java.io.IOException;
import java.lang.ref.Reference;
import java.lang.ref.SoftReference;
import java.lang.ref.WeakReference;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Enumeration;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Vector;
import java.util.WeakHashMap;
import java.util.logging.Level;
import java.util.logging.Logger;

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

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

    /**
     * Size of the message queue provided by virtual messengers.
     */
    private final static int DEFAULT_MESSAGE_QUEUE_SIZE = 20;

    /**
     * If {@code true} then the parent endpoint may be used for acquiring
     * messengers and for registering listeners.
     */
    private final static boolean DEFAULT_USE_PARENT_ENDPOINT = true;

    EndpointServiceMonitor endpointServiceMonitor;

    /**
     * the EndpointMeter
     */
    private EndpointMeter endpointMeter;
    private PropagationMeter propagationMeter;

    /**
     * If {@code true} then this service has been initialized.
     */
    private boolean initialized = false;

    /**
     * tunable: the virtual messenger queue size
     */
    private int vmQueueSize = DEFAULT_MESSAGE_QUEUE_SIZE;

    private PeerGroup group = null;
    private ID assignedID = null;
    private ModuleImplAdvertisement implAdvertisement = null;

    private String localPeerId = null;
    private boolean useParentEndpoint = DEFAULT_USE_PARENT_ENDPOINT;
    private EndpointService parentEndpoint = null;
    private String myServiceName = 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.
     */
    private final Collection<MessageTransport> messageTransports = new HashSet<MessageTransport>();

    /**
     * Passive listeners for messengers. Three priorities, so far.
     */
    private final Collection[] passiveMessengerListeners = {
            Collections.synchronizedList(new ArrayList<MessengerEventListener>()),
            Collections.synchronizedList(new ArrayList<MessengerEventListener>()),
            Collections.synchronizedList(new ArrayList<MessengerEventListener>())
    };

    /**
     * The set of listener managed by this instance of the endpoint svc.
     */
    private final Map<String, EndpointListener> incomingMessageListeners = new HashMap<String, EndpointListener>(16);

    /**
     * The set of shared transport messengers currently ready for use.
     */
    private final Map<EndpointAddress, Reference<Messenger>> messengerMap = new WeakHashMap<EndpointAddress, Reference<Messenger>>(32);
    /**
     * The set of shared transport messengers currently ready for use.
     */
    private final Map<EndpointAddress, Reference<Messenger>> directMessengerMap = new WeakHashMap<EndpointAddress, Reference<Messenger>>(32);

    /**
     * The filter listeners.
     * <p/>
     * 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.
     */
    private final Collection<FilterListenerAndMask> incomingFilterListeners = new ArrayList<FilterListenerAndMask>();
    private final Collection<FilterListenerAndMask> outgoingFilterListeners = new ArrayList<FilterListenerAndMask>();

    /**
     * Holder for a filter listener and its conditions
     */
    private static class FilterListenerAndMask {
        final String namespace;
        final String name;
        final MessageFilterListener listener;

        public FilterListenerAndMask(MessageFilterListener listener, String namespace, String name) {
            this.namespace = namespace;
            this.name = name;
            this.listener = listener;
        }

        @Override
        public boolean equals(Object target) {
            if (this == target) {
                return true;
            }

            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 result;
            }

            return false;
        }

        /**
         * {@inheritDoc}
         * <p/>
         * Added to make PMD shut up....
         */
        @Override
        public int hashCode() {
            return System.identityHashCode(this);
        }
    }


    /**
     * 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.
         * <p/>
         * 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 vmQueueSize        queue size
         * @param destination        destination who messages should be addressed to
         * @param logicalDestination logical destination