wireformatmessagebinary.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.document.MimeMediaType;
import net.jxta.endpoint.ByteArrayMessageElement;
import net.jxta.endpoint.Message;
import net.jxta.endpoint.MessageElement;
import net.jxta.endpoint.WireFormatMessage;
import net.jxta.endpoint.WireFormatMessageFactory;
import net.jxta.util.LimitInputStream;
import java.util.logging.Level;
import net.jxta.logging.Logging;
import java.util.logging.Logger;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.EOFException;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.io.SequenceInputStream;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.nio.ByteBuffer;
import java.text.MessageFormat;
import java.util.Arrays;


/**
 * A Wire Format Message which encodes the message into MIME Type
 * "application/x-jxta-msg".
 * <p/>
 * <p/>This implementation does nothing with encodings.
 * <p/>
 * <p/>This implementation does not use any MIME parameters attached to the
 * requesting mime type.
 *
 * @see net.jxta.endpoint.WireFormatMessageFactory
 * @see <a href="https://jxta-spec.dev.java.net/nonav/JXTAProtocols.html#msgs-fmts-jbm" target="_blank">JXTA Protocols Specification : Binary Message Format</a>
 */
public class WireFormatMessageBinary implements WireFormatMessage {

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

    // Flag bits
    protected static final byte HAS_TYPE = 0x01;
    protected static final byte HAS_ENCODING = 0x02;
    protected static final byte HAS_SIGNATURE = 0x04;

    protected static final int MESSAGE_VERSION = 0;

    /**
     * Our Mime Media Type(s)
     */
    private static final MimeMediaType[] myTypes = {
        MimeMediaType.valueOf("application/x-jxta-msg") };

    /**
     * These are the content encodings we support.
     */
    private static final MimeMediaType[] myContentEncodings = {

        // we support raw binary!
        null
    };

    /**
     * Our instantiator for the factory.
     */
    public static final WireFormatMessageFactory.Instantiator INSTANTIATOR = new Instantiator();

    /**
     * Our instantiator.
     */
    static class Instantiator implements WireFormatMessageFactory.Instantiator {

        /**
         * Creates new WireFormatMessageBinary Instantiator
         */
        public Instantiator() {}

        /**
         * {@inheritDoc}
         */
        public MimeMediaType[] getSupportedMimeTypes() {
            return myTypes;
        }

        /**
         * {@inheritDoc}
         */
        public MimeMediaType[] getSupportedContentEncodings() {
            return myContentEncodings;
        }

        /**
         * {@inheritDoc}
         */
        public Message fromWire(InputStream is, MimeMediaType type, MimeMediaType contentEncoding) throws IOException {
            // FIXME 20020504 [e-mail]  Ignores type and contentEncoding completely.
            Message msg = new Message();

            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.fine("Reading " + msg + " from " + is);
            }

            DataInputStream dis = new DataInputStream(is);

            HashMap idToNamespace = readHeader(dis);

            int elementCnt = dis.readShort();

            if (Logging.SHOW_FINER && LOG.isLoggable(Level.FINER)) {
                LOG.finer("Message element count " + elementCnt + " from " + is);
            }

            int eachElement = 0;

            do {
                if (Logging.SHOW_FINER && LOG.isLoggable(Level.FINER)) {
                    LOG.finer("Read element " + eachElement + " of " + elementCnt + " from " + is + " for " + msg);
                }

                Object[] anElement;

                try {
                    anElement = readMessageElement(dis, is);
                } catch (IOException failed) {
                    if (Logging.SHOW_SEVERE && LOG.isLoggable(Level.SEVERE)) {
                        LOG.log(Level.SEVERE
                                ,
                                "Failure reading element " + eachElement + " of " + elementCnt + " from " + is + " for " + msg
                                ,
                                failed);
                    }

                    throw failed;
                }

                if (null == anElement) {
                    break;
                }

                String namespace = (String) idToNamespace.get(anElement[0]);

                if (null == namespace) {
                    if (Logging.SHOW_SEVERE && LOG.isLoggable(Level.SEVERE)) {
                        LOG.severe("Element identified a namespace which was not defined for this message.");
                    }

                    throw new IOException("Element identified a namespace which was not defined for this message.");
                }

                msg.addMessageElement(namespace, (MessageElement) anElement[1]);
                eachElement++;

                if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINER)) {
                    LOG.finer(
                            "Add element (name=\'" + ((MessageElement) anElement[1]).getElementName() + "\') #" + eachElement
                            + " of #" + elementCnt + " elements from " + dis.toString());
                }
            } while (((0 == elementCnt) || (eachElement < elementCnt)));

            if ((elementCnt != 0) && (eachElement != elementCnt)) {
                throw new IOException("Found wrong number of elements in message.");
            }

            return msg;
        }

        public Message fromBuffer(ByteBuffer buffer, MimeMediaType type, MimeMediaType contentEncoding) throws IOException {
            // FIXME 20020504 [e-mail]  Ignores type and contentEncoding completely.
            Message msg = new Message();

            if (Logging.SHOW_FINE && LOG.isLoggable(Level.FINE)) {
                LOG.fine("Reading " + msg + " from " + buffer);
            }

            HashMap idToNamespace = readHeader(buffer);

            int elementCnt = buffer.getShort();

            if (Logging.SHOW_FINER && LOG.isLoggable(Level.FINER)) {
                LOG.finer("Message element count " + elementCnt + " from " + buffer);
            }

            int eachElement = 0;

            do {
                if (Logging.SHOW_FINER && LOG.isLoggable(Level.FINER)) {
                    LOG.finer("Read element " + eachElement + " of " + elementCnt + " from " + buffer + " for " + msg);
                }

                Object[] anElement;

                try {
                    anElement = readMessageElement(buffer);

                    if (Logging.SHOW_FINER && LOG.isLoggable(Level.FINER)) {
                        LOG.finer(
                                MessageFormat.format("Read element of size {0}, [{1}] {2}", anElement.length, anElement.toString()
                                ,
                                buffer.toString()));
                    }
                } catch (IOException failed) {
                    if (Logging.SHOW_SEVERE && LOG.isLoggable(Level.SEVERE)) {
                        LOG.log(Level.SEVERE
                                ,
                                "Failure reading element " + eachElement + " of " + elementCnt + " from " + buffer + " for " + msg
                                ,
                                failed);
                    }
                    throw failed;
                }

                if (null == anElement) {
                    break;
                }

                String namespace = (String) idToNamespace.get(anElement[0]);

                if (null == namespace) {
                    if (Logging.SHOW_SEVERE && LOG.isLoggable(Level.SEVERE)) {
                        LOG.severe("Element identified a namespace which was not defined for this message.");
                    }
                    throw new IOException("Element identified a namespace which was not defined for this message.");
                }

                msg.addMessageElement(namespace, (MessageElement) anElement[1]);
                eachElement++;

                if (Logging.SHOW_FINER && LOG.isLoggable(Level.FINER)) {
                    LOG.finer(
                            "Add element (name=\'" + ((MessageElement) anElement[1]).getElementName() + "\') #" + eachElement
                            + " of #" + elementCnt + " elements from " + buffer.toString());
                }
            } while (((0 == elementCnt) || (eachElement < elementCnt)));

            if ((elementCnt != 0) && (eachElement != elementCnt)) {
                throw new IOException("Found wrong number of elements in message.");
            }

            return msg;
        }

        /**
         * {@inheritDoc}
         */
        public WireFormatMessage toWire(Message msg, MimeMediaType type, MimeMediaType[] preferedContentEncoding) {
            try {
                return new WireFormatMessageBinary(msg, type, preferedContentEncoding);
            } catch (IOException caught) {
                throw new IllegalStateException("Could not build wire format for message due to " + caught.getMessage());
            }
        }

        /**
         * Read in a message header from the provided data stream.
         *
         * @param dis the data stream to read from
         * @return hashmap containing the namespace id to namespace values
         * @throws IOException if EOF or other IOException is encountered
         *                     during the reading of the header.
         */
        private static HashMap readHeader(DataInputStream dis) throws IOException {
            // Read message signature
            char[] msgsig = new char[4];

            try {
                msgsig[0] = (char) dis.readByte();
            } catch (EOFException failed) {
                if (Logging.SHOW_FINER && LOG.isLoggable(Level.FINER)) {
                    LOG.log(Level.FINER, "EOF reading message at first byte of header.", failed);
                }

                throw failed;
            }
            msgsig[1] = (char) dis.readByte();
            msgsig[2] = (char) dis.readByte();
            msgsig[3] = (char) dis.readByte();

            if (msgsig[0] != 'j' || msgsig[1] != 'x' || msgsig[2] != 'm' || msgsig[3] != 'g') {
                IOException failure = new IOException(
                        "Not a message (incorrect signature '" + msgsig[0] + msgsig[1] + msgsig[2] + msgsig[3] + "') ");

                if (Logging.SHOW_SEVERE && LOG.isLoggable(Level.SEVERE)) {
                    LOG.severe(failure.toString());
                }

                throw failure;
            }

            // Message version
            if (dis.readByte() != MESSAGE_VERSION) {
                IOException failure = new IOException("Message not version " + MESSAGE_VERSION);

                if (Logging.SHOW_SEVERE && LOG.isLoggable(Level.SEVERE)) {
                    LOG.log(Level.SEVERE, failure.getMessage(), failure);
                }

                throw failure;
            }

            int namespaceCnt = dis.readShort();

            if (namespaceCnt > 253) {
                IOException failure = new IOException("Message contains too many namespaces (>253)");

                if (Logging.SHOW_SEVERE && LOG.isLoggable(Level.SEVERE)) {
                    LOG.log(Level.SEVERE, failure.getMessage(), failure);
                }

                throw failure;
            }

            HashMap<Integer, String> id2namespace = new HashMap<Integer, String>(2 + namespaceCnt);

            id2namespace.put(0, "");
            id2namespace.put(1, "jxta");

            int id = 2;

            for (int i = 0; i < namespaceCnt; ++i) {
                try {
                    String namespace = readString(dis);

                    id2namespace.put(id++, namespace);
                } catch (IOException caught) {
                    if (Logging.SHOW_WARNING && LOG.isLoggable(Level.WARNING)) {
                        LOG.log(Level.WARNING, "Error Processing namespace", caught);
                    }
                    throw caught;
                }
            }

            if (Logging.SHOW_FINER && LOG.isLoggable(Level.FINER)) {
                LOG.finer("Read Message Header with " + (namespaceCnt + 2) + " namespaces from " + dis.toString());
            }

            return id2namespace;
        }

        /**
         * Read in a message header from the provided data stream.
         *
         * @param buffer the data buffer to read from
         * @return hashmap containing the namespace id to namespace values
         * @throws IOException if EOF or other IOException is encountered
         *                     during the reading of the header.
         */
        private static HashMap readHeader(ByteBuffer buffer) throws IOException {
            // Read message signature
            char[] msgsig = new char[4];

            msgsig[0] = (char) buffer.get();
            msgsig[1] = (char) buffer.get();
            msgsig[2] = (char) buffer.get();
            msgsig[3] = (char) buffer.get();

            if (msgsig[0] != 'j' || msgsig[1] != 'x' || msgsig[2] != 'm' || msgsig[3] != 'g') {
                IOException failure = new IOException(
                        "Not a message (incorrect signature '" + msgsig[0] + msgsig[1] + msgsig[2] + msgsig[3] + "') ");