wireformatmessagebinary.java

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

            id2namespace.put(new Integer(1), "jxta");
            
            int id=2;
            for(int i=0; i<namespaceCnt; ++i) {
                try {
                    String namespace = readString(dis);
                    id2namespace.put(new Integer(id++), namespace);
                } catch ( IOException caught ) {
                    if (LOG.isEnabledFor(Level.WARN)) {
                        LOG.warn( "Error Processing namespace", caught );
                    }
                    throw caught;
                }
            }
            
            if ( LOG.isEnabledFor(Level.DEBUG) ) {
                LOG.debug( "Read Message Header with " + (namespaceCnt + 2) + " namespaces from " + dis.toString() );
            }
            
            return id2namespace;
        }
        
        /**
         *  Read in a message element from the provided data stream.
         *
         *  @param dis  the data stream to read from
         *  @return object array containing two objects, index[0] contains an
         *  Integer which identifies the namespace to which this element belongs
         *  and index[1] contains a MessageElement. If null is returned then
         *  the DataInputStream reached EOF before reading the first byte of the
         *  element.
         *  @throws IOException if EOF or other IOException is encountered
         *  during the reading of the element.
         **/
        private Object [] readMessageElement( DataInputStream dis, InputStream is ) throws IOException {
            // Read message signature
            char [] elsig = new char[4];
            
            // if we EOF before the first byte, return null. EOF anywhere else
            // and its an error.
            try {
                elsig[0] = (char) dis.readByte();
            } catch( EOFException allDone ) {
                return null;
            }
            
            elsig[1] = (char) dis.readByte();
            elsig[2] = (char) dis.readByte();
            elsig[3] = (char) dis.readByte();
            
            if ( elsig[0] != 'j' || elsig[1] != 'x' || elsig[2] != 'e' || elsig[3] != 'l') {
                IOException failure = new IOException( "Not a message element (incorrect signature '" +
                elsig[0] + elsig[1] + elsig[2] + elsig[3]
                + "') "  );
                
                if ( LOG.isEnabledFor(Level.ERROR) ) {
                    LOG.error( failure, failure );
                }
                
                throw failure;
            }
            
            // Namespace id
            int nsid = dis.readByte();
            
            // flags
            byte flags = dis.readByte();
            
            // Name
            String name = readString(dis);
            
            // Mime type
            MimeMediaType type = null;
            if ((flags & HAS_TYPE) != 0) {
                String typeString = readString(dis);
                try {
                    type = new MimeMediaType(typeString);
                } catch ( IllegalArgumentException uhoh ) {
                    throw new IOException( "Bad MIME type in message element header : " + uhoh.getMessage() );
                }
            } else {
                type = MimeMediaType.AOS;
            }
            
            int dataLen = dis.readInt();
            
            if ( LOG.isEnabledFor(Level.TRACE) ) {
                LOG.trace(  "element : nsid = " + nsid +
                " name = '" + name +
                "' type = '" + type +
                "' flags = " + Integer.toBinaryString( flags ) +
                " datalen = " + dataLen );
            }
            
            Object [] res = new Object [2];
            
            res[0] = new Integer( nsid & 0x000000FF );
            
            byte[] value = null;
            Message submsg = null;
            
            // Value
            if ( type.equalsIngoringParams( myTypes [0] ) ) {
                InputStream subis = new LimitInputStream( is, dataLen );
                
                submsg = WireFormatMessageFactory.fromWire( subis, type, null );
            } else {
                if( dataLen > Integer.MAX_VALUE ) {
                    if ( LOG.isEnabledFor(Level.ERROR) ) {
                        LOG.error( "WireFormatMessageBinary does not support elements longer than 2GB" );
                    }
                    
                    throw new IllegalStateException( "WireFormatMessageBinary does not support elements longer than 2GB" );
                }
                
                value = new byte[dataLen];

                String mayFail = null;
                if (LOG.isEnabledFor(Level.WARN)) {
                    mayFail = is.toString();
                }
                
                try {
                    dis.readFully( value );
                } catch( EOFException failed ) {
                    if (LOG.isEnabledFor(Level.WARN)) {
                        LOG.error( "had tried to read " + dataLen + " from " + mayFail + " which is now " + is );
                    }
                    throw failed;
                }
            }
            
            MessageElement sig = null;
            if ((flags & HAS_SIGNATURE) != 0) {
                Object [] sigRes = readMessageElement( dis, is );
                sig = (MessageElement) sigRes[1];
            }
            
            if( null != value ) {
                res[1] = new ByteArrayMessageElement( name, type, value, sig );
            } else {
                res[1] = new JxtaMessageMessageElement( name, type, submsg, sig );
            }
            
            return res;
        }
        
        /**
         *  Read and construct a string from the data stream.
         *
         *  @param dis the stream to read from
         *  @return the String which was read.
         *  @throws IOException if EOF or other IOException is encountered
         *  during the reading of the string.
         **/
        private static String readString(DataInputStream dis) throws IOException {
            int len = dis.readShort();
            if (len < 0) {
                throw new IOException("Bad string length in message");
            }
            
            byte[] bytes = new byte[len];
            dis.readFully(bytes);
            return new String( bytes, "UTF8");
        }
    };
    
    /**
     *  Internal representation for a binary format wire message. Implemented
     *  as an inner class to allow content encodings to be easily mapped on
     *  top of the streams this class produces.
     **/
    static class binaryMessageProxy implements Document {
        final Message message;
        
        final MimeMediaType type;
        
        final List elements = new ArrayList();
        
        final Map namespaceIDs = new HashMap();
        
        final List namespaces = new ArrayList();
        
        byte [] header;
        
        binaryMessageProxy( Message msg, MimeMediaType type ) throws IOException {
            message = msg;
            
            this.type = type; // we may generate different content based upon the type.
            
            assignNamespaceIds();
            
            // build the element proxies
            Message.ElementIterator eachElement = message.getMessageElements();
            
            while( eachElement.hasNext() ) {
                MessageElement anElement = (MessageElement) eachElement.next();
                byte namespaceid = ((Integer)namespaceIDs.get( eachElement.getNamespace() )).byteValue();
                
                elements.add( new binaryElementProxy( namespaceid, anElement ) );
            }
            
            buildHeader();
        }
        
        /**
         * {@inheritDoc}
         **/
        public String getFileExtension() {
            return "???";
        }
        
        /**
         * {@inheritDoc}
         **/
        public MimeMediaType getMimeType() {
            return type;
        }
        
        /**
         * {@inheritDoc}
         **/
        public InputStream getStream() throws IOException {
            
            List streamParts = new ArrayList();
            
            streamParts.add( new ByteArrayInputStream( header ) );
            
            Iterator eachElement = elements.listIterator();
            
            while( eachElement.hasNext() ) {
                binaryElementProxy anElement = (binaryElementProxy) eachElement.next();
                streamParts.add( anElement.getStream() );
            }
            
            InputStream theStream = new SequenceInputStream( Collections.enumeration( streamParts ) );
            
            if ( LOG.isEnabledFor(Level.DEBUG) ) {
                LOG.debug( "Returning " + theStream.getClass().getName() + "@" + System.identityHashCode(theStream) + " for " + message );
            }
            
            return theStream;
        }
        
        /**
         * {@inheritDoc}
         **/
        public void sendToStream( OutputStream sendTo ) throws IOException {
            
            if ( LOG.isEnabledFor(Level.DEBUG) ) {
                LOG.debug( "Sending " + message + " to " + sendTo.getClass().getName() + "@" + System.identityHashCode(sendTo) );
            }
            
            sendTo.write( header );
            
            Iterator eachElement = elements.listIterator();
            
            while( eachElement.hasNext() ) {
                binaryElementProxy anElement = (binaryElementProxy) eachElement.next();
                anElement.sendToStream( sendTo );
            }
        }
        
        /**
         * {@inheritDoc}
         **/
        public long getByteLength() {
            long size = 0;
            
            size += header.length;
            Iterator eachElement = elements.iterator();
            while( eachElement.hasNext() ) {
                binaryElementProxy anElement = (binaryElementProxy) eachElement.next();
                
                size += anElement.getByteLength();
            }
            
            return size;
        }
        
        /**
         *  Scans the source message to build a HashMap of the namespaces used
         *  in the message and assign and id to each namespace.
         */
        private void assignNamespaceIds( ) {
            int id = 0;
            Iterator namespaces = message.getMessageNamespaces();
            
            // insert the predefined namespaces.
            namespaceIDs.put("", new Integer(id++) );
            this.namespaces.add( "" );
            namespaceIDs.put("jxta", new Integer(id++) );
            this.namespaces.add( "jxta" );
            
            // insert items in the vector if they are not found in the map
            while( namespaces.hasNext() ) {
                String namespace = (String) namespaces.next();
                if (namespaceIDs.get(namespace) == null) {
                    namespaceIDs.put(namespace, new Integer(id++));
                    this.namespaces.add(namespace);
                }
            }
            
            if( id >= 256 ) {