message.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.endpoint;

import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.PrintWriter;
import java.io.Serializable;
import java.io.StringWriter;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.ListIterator;
import java.util.Map;

import java.io.IOException;
import java.util.ConcurrentModificationException;
import java.util.NoSuchElementException;
import java.util.concurrent.atomic.AtomicInteger;

import java.util.logging.Level;
import net.jxta.logging.Logging;
import java.util.logging.Logger;

import net.jxta.document.MimeMediaType;
import net.jxta.util.AbstractSimpleSelectable;
import net.jxta.util.SimpleSelectable;

import net.jxta.impl.id.UUID.UUID;
import net.jxta.impl.id.UUID.UUIDFactory;

/**
 * Messages are abstract containers for protocol messages within JXTA. Services
 * and applications are expected to use Messages as the basis for any protocols
 * implemented within JXTA. Messages are exchanged through the
 * {@link net.jxta.endpoint.EndpointService} or
 * {@link net.jxta.pipe.PipeService}.
 * <p/>
 * A Message is composed of an ordered list of zero or more
 * {@link net.jxta.endpoint.MessageElement MessageElements}. Each
 * {@link net.jxta.endpoint.MessageElement} is associated with a namespace at
 * the time it is added to the message. Duplicate
 * {@link net.jxta.endpoint.MessageElement MessageElements} are permitted.
 * <p/>
 * <b>Messages are not synchronized. All of the iterators returned by this
 * implementation are "fail-fast". Concurrent modification of messages from
 * multiple threads will produce unexpected results and
 * {@code ConcurrentModificationException}.</b>
 *
 * @see net.jxta.endpoint.MessageElement
 * @see net.jxta.endpoint.EndpointAddress
 * @see net.jxta.endpoint.EndpointService
 * @see net.jxta.pipe.InputPipe
 * @see net.jxta.pipe.OutputPipe
 * @see net.jxta.pipe.PipeService
 */
public class Message extends AbstractSimpleSelectable implements Serializable {

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

    /**
     * Magic value for this format of serialization version.
     */
    private static final long serialVersionUID = 3418026921074097757L;

    /**
     * If {@code true}, then modification logging be activated. This is a very
     * expensive option as it causes a stack crawl to be captured for every
     * message modification.
     * <p/>
     * To enable modification tracking, set to {@code true} and recompile.
     */
    protected static final boolean LOG_MODIFICATIONS = false;

    /**
     * If {@code true}, then a special tracking element is added to every
     * message. This provides the ability to follow messages throughout the
     * network. If a message has a tracking element then it will be used in
     * the {@code toString()} representation.
     * <p/>
     * The element is currently named "Tracking UUID" and is stored in the
     * "jxta" namespace. The element contains an IETF version 1 UUID in string
     * form.
     * <p/>
     * To enable addition of a tracking element to every message, set the
     * Java System Property {@code net.jxta.endpoint.Message.globalTracking} to
     * {@code true} and restart your JXTA application.
     *
     * @see java.lang.System#setProperty(String,String)
     */
    protected static final boolean GLOBAL_TRACKING_ELEMENT =
            Boolean.getBoolean(Message.class.getName() + ".globalTracking");

    /**
     * Incremented for each standalone message instance. {@see #lineage} for
     * information about how message numbers can be used.
     */
    private static transient AtomicInteger messagenumber = new AtomicInteger(1);

    /**
     * This string identifies the namespace which is assumed when calls are
     * made that don't include a namespace specification.
     */
    protected final String defaultNamespace;

    /**
     * the namespaces in this message and the elements in each.
     */
    protected transient Map<String, List<MessageElement>> namespaces = new HashMap<String, List<MessageElement>>();

    /**
     * List of the elements.
     */
    protected transient List<element> elements = new ArrayList<element>();

    /**
     * Message properties HashMap
     */
    protected transient Map<Object, Object> properties = Collections.synchronizedMap(new HashMap<Object, Object>());

    /**
     * A list of {@link java.lang.Integer} which details the lineage (history
     * of cloning) that produced this message. This message's number is index
     * 0, all of the ancestors are in order at higher indexes.
     * <p/>
     * Message numbers are not part of the message content and are only
     * stored locally. The are useful for following messages throughout their
     * lifetime and is normally shown as part of the <tt>toString()</tt>
     * display for Messages.
     */
    protected transient List<Integer> lineage = new ArrayList<Integer>();

    /**
     * Modification count of this message. Can be used to detect message being
     * concurrently modified when message is shared.
     * <p/>
     * The modification count is part of the {@code toString()} display for
     * Messages.
     */
    protected transient volatile int modCount = 0;

    /**
     * cached aggregate size of all the member elements. Used by
     * {@link #getByteLength()}
     */
    protected transient long cachedByteLength = 0;

    /**
     * modcount at the time the message length was last calculated. Used by
     * {@link #getByteLength()}
     */
    protected transient int cachedByteLengthModCount = -1;

    /**
     * If <tt>true</tt> then the message is modifiable. This is primarily
     * intended as a diagnostic tool for detecting concurrent modification.
     *
     * @deprecated You really should not depend on this feature.
     */
    @Deprecated
    public boolean modifiable = true;

    /**
     * If {@code LOG_MODIFICATIONS} is {@code true} then this will contain
     * the history of modifications this message.
     * <p/>
     * <ul>
     * <li>Values are {@link java.lang.Throwable} with the description
     * field formatted as <code>timeInAbsoluteMillis : threadName</code>.
     * </li>
     * </ul>
     */
    protected transient List<Throwable> modHistory;

    /**
     * A ListIterator for MessageElements which also provides the ability to
     * determine the namespace of the current message element. Message Elements
     * are iterated in the order in which they were added to the Message.
     * <p/>
     * This Iterator returned is not synchronized with the message. If you
     * modify the state of the Message, the iterator will throw
     * ConcurrentModificationException when {@code next()} or
     * {@code previous()} is called.
     */
    public class ElementIterator implements ListIterator<MessageElement> {

        /**
         * The elements being iterated.
         */
        ListIterator<element> list;

        /**
         * The current element
         */
        element current = null;

        /**
         * The modCount at the time when the iterator was created.
         */
        transient int origModCount;

        /**
         * Intialize the iterator from a list iterator.
         *
         * @param list The ListIterator we are managing.
         */
        ElementIterator(ListIterator<element> list) {
            origModCount = Message.this.getMessageModCount();
            this.list = list;
        }

        /**
         * {@inheritDoc}
         */
        public boolean hasNext() {
            if (origModCount != Message.this.getMessageModCount()) {
                RuntimeException failure = new ConcurrentModificationException(
                        Message.this + " concurrently modified. Iterator was made at mod " + origModCount);

                if (Logging.SHOW_SEVERE && LOG.isLoggable(Level.SEVERE)) {
                    LOG.log(Level.SEVERE,
                            Message.this + " concurrently modified. iterator mod=" + origModCount + " current mod="
                            + Message.this.getMessageModCount() + "\n" + getMessageModHistory(),
                            failure);
                }
                throw failure;
            }
            return list.hasNext();
        }

        /**
         * {@inheritDoc}
         */
        public MessageElement next() {
            if (origModCount != Message.this.getMessageModCount()) {
                RuntimeException failure = new ConcurrentModificationException(
                        Message.this + " concurrently modified. Iterator was made at mod " + origModCount);

                if (Logging.SHOW_SEVERE && LOG.isLoggable(Level.SEVERE)) {
                    LOG.log(Level.SEVERE,
                            Message.this + " concurrently modified. iterator mod=" + origModCount + " current mod="
                            + Message.this.getMessageModCount() + "\n" + getMessageModHistory(),
                            failure);
                }
                throw failure;
            }

            current = list.next();
            return current.element;
        }

        /**
         * {@inheritDoc}
         */
        public int nextIndex() {
            return list.nextIndex();
        }

        /**
         * {@inheritDoc}
         */
        public boolean hasPrevious() {
            if (origModCount != Message.this.getMessageModCount()) {
                RuntimeException failure = new ConcurrentModificationException(
                        Message.this + " concurrently modified. Iterator was made at mod " + origModCount);

                if (Logging.SHOW_SEVERE && LOG.isLoggable(Level.SEVERE)) {
                    LOG.log(Level.SEVERE,
                            Message.this + " concurrently modified. iterator mod=" + origModCount + " current mod="
                            + Message.this.getMessageModCount() + "\n" + getMessageModHistory(),
                            failure);
                }
                throw failure;
            }

            return list.hasPrevious();
        }

        /**
         * {@inheritDoc}
         */
        public MessageElement previous() {
            if (origModCount != Message.this.getMessageModCount()) {
                RuntimeException failure = new ConcurrentModificationException(
                        Message.this + " concurrently modified. Iterator was made at mod " + origModCount);

                if (Logging.SHOW_SEVERE && LOG.isLoggable(Level.SEVERE)) {
                    LOG.log(Level.SEVERE,
                            Message.this + " concurrently modified. iterator mod=" + origModCount + " current mod="
                            + Message.this.getMessageModCount() + "\n" + getMessageModHistory(),
                            failure);
                }
                throw failure;
            }

            current = list.previous();
            return current.element;
        }

        /**
         * {@inheritDoc}
         */
        public int previousIndex() {
            return list.previousIndex();