| // Licensed to the Apache Software Foundation (ASF) under one |
| // or more contributor license agreements. See the NOTICE file |
| // distributed with this work for additional information |
| // regarding copyright ownership. The ASF licenses this file |
| // to you under the Apache License, Version 2.0 (the |
| // "License"); you may not use this file except in compliance |
| // with the License. You may obtain a copy of the License at |
| // |
| // http://www.apache.org/licenses/LICENSE-2.0 |
| // |
| // Unless required by applicable law or agreed to in writing, |
| // software distributed under the License is distributed on an |
| // "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| // KIND, either express or implied. See the License for the |
| // specific language governing permissions and limitations |
| // under the License. |
| /* Took the basic code from Axis 1.2 and modified to fit into the cloud code base */ |
| |
| package com.cloud.bridge.io; |
| |
| import java.io.IOException; |
| import java.io.InputStream; |
| import java.io.FilterInputStream; |
| import org.apache.log4j.Logger; |
| |
| |
| /** |
| * This class takes the input stream and turns it multiple streams. |
| DIME version 0 format |
| <pre> |
| 0 1 2 3 |
| 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 |
| +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ --- |
| | VERSION |B|E|C| TYPE_T| OPT_T | OPTIONS_LENGTH | A |
| +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ |
| | ID_LENGTH | TYPE_LENGTH | Always present 12 bytes |
| +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ even on chunked data. |
| | DATA_LENGTH | V |
| +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ --- |
| | / |
| / OPTIONS + PADDING / |
| / (absent for version 0) | |
| +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ |
| | / |
| / ID + PADDING / |
| / | |
| +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ |
| | / |
| / TYPE + PADDING / |
| / | |
| +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ |
| | / |
| / DATA + PADDING / |
| / | |
| +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+ |
| </pre> |
| * This implementation of input stream does not support marking operations. |
| * |
| * Incoming data is DIME encoded when its MIME type is "application/dime". |
| * Then use this class to pull out 2 streams: |
| * (1) The first stream is the SOAP request, |
| * (2) The second stream is a chunked attachment (e.g., a file to store) |
| * |
| * The DIME format is defined at this reference: |
| * http://msdn.microsoft.com/en-us/library/aa480488.aspx |
| * |
| * @author Rick Rineholt |
| */ |
| public class DimeDelimitedInputStream extends FilterInputStream |
| { |
| protected final static Logger logger = Logger.getLogger(DimeDelimitedInputStream.class); |
| |
| InputStream is = null; //The source input stream. |
| boolean closed = true; //The stream has been closed. |
| boolean theEnd = false; //There are no more streams left. |
| boolean moreChunks = false; //More chunks are a coming! |
| boolean MB = false; //Message begin flag |
| boolean ME = false; //Message end flag |
| String type = null; // |
| String id = null; // |
| String tnf = null; //DIME type format |
| long recordLength = 0L; //length of the current record. |
| long bytesRead = 0L; //How many bytes of the current record have been read. |
| int dataPadLength = 0; //How many pad bytes there are. |
| |
| protected int streamNo = 0; |
| protected IOException streamInError = null; |
| |
| private static byte[] trash = new byte[4]; |
| protected static int streamCount = 0; //number of streams produced. |
| |
| protected static synchronized int newStreamNo() |
| { |
| logger.debug( "streamNo " + (streamCount + 1)); |
| return ++streamCount; |
| } |
| |
| |
| /** |
| * There can be multiple streams in a DIME encoding. For example, the first |
| * stream can be a SOAP message, and the second stream a binary attachment (e.g., |
| * a file). During reading after an EOF is returned, this function should be |
| * called to see if there is another stream following the last. |
| * |
| * @return the dime delimited stream, null if there are no more streams |
| * @throws IOException if there was an error loading the data for the next stream |
| */ |
| public synchronized DimeDelimitedInputStream getNextStream() throws IOException |
| { |
| if (null != streamInError) throw streamInError; |
| if (theEnd) return null; |
| |
| //Each Stream must be read in succession |
| if (bytesRead < recordLength || moreChunks) |
| throw new RuntimeException("attach.dimeReadFullyError"); |
| |
| dataPadLength -= readPad(dataPadLength); |
| |
| //Create an new dime stream that comes after this one. |
| return new DimeDelimitedInputStream( this.is ); |
| } |
| |
| |
| /** |
| * Create a new dime stream. |
| * |
| * @param is the <code>InputStream</code> to wrap |
| * @throws IOException if anything goes wrong |
| */ |
| public DimeDelimitedInputStream( InputStream is ) throws IOException |
| { |
| super(null); |
| streamNo = newStreamNo(); |
| closed = false; |
| this.is = is; |
| readHeader( false ); |
| } |
| |
| |
| /** |
| * Make sure to skip the pad which appear in several parts of a DIME message. |
| * @param size |
| * @return |
| * @throws IOException |
| */ |
| private final int readPad( int size ) throws IOException |
| { |
| if (0 == size) return 0; |
| int read = readFromStream(trash, 0, size); |
| |
| if (size != read) |
| { |
| streamInError = new IOException("attach.dimeNotPaddedCorrectly"); |
| throw streamInError; |
| } |
| return read; |
| } |
| |
| |
| private final int readFromStream( byte[] b ) throws IOException |
| { |
| return readFromStream( b, 0, b.length ); |
| } |
| |
| private final int readFromStream( byte[] b, int start, int length ) throws IOException |
| { |
| int br = 0; |
| int brTotal = 0; |
| |
| if (length == 0) return 0; |
| |
| do |
| { try |
| { |
| br = is.read( b, brTotal + start, length - brTotal ); |
| } |
| catch (IOException e) |
| { |
| streamInError = e; |
| throw e; |
| } |
| if (br > 0) brTotal += br; |
| } |
| while( br > -1 && brTotal < length ); |
| |
| return br > -1 ? brTotal : br; |
| } |
| |
| |
| /** |
| * Get the id for this stream part. |
| * @return the id; |
| */ |
| public String getContentId() |
| { |
| return id; |
| } |
| |
| public String getDimeTypeNameFormat() |
| { |
| return tnf; |
| } |
| |
| /** |
| * Get the type, as read from the header. |
| * @return the type of this dime |
| */ |
| public String getType() |
| { |
| return type; |
| } |
| |
| |
| /** |
| * Read from the DIME stream. |
| * |
| * @param b is the array to read into. |
| * @param off is the offset |
| * @return the number of bytes read. -1 if endof stream |
| * @throws IOException if data could not be read from the stream |
| */ |
| public synchronized int read( byte[] b, int off, int len ) throws IOException |
| { |
| if (closed) |
| { |
| dataPadLength -= readPad(dataPadLength); |
| throw new IOException( "streamClosed" ); |
| } |
| return _read( b, off, len ); |
| } |
| |
| |
| protected int _read( byte[] b, int off, int len ) throws IOException |
| { |
| int totalbytesread = 0; |
| int bytes2read = 0; |
| |
| if (len < 0) |
| throw new IllegalArgumentException( "attach.readLengthError" + len ); |
| |
| if (off < 0) |
| throw new IllegalArgumentException( "attach.readOffsetError" + off ); |
| |
| if (b == null) |
| throw new IllegalArgumentException( "attach.readArrayNullError" ); |
| |
| if (b.length < off + len) |
| throw new IllegalArgumentException("attach.readArraySizeError " + b.length + " " + len + " " + off ); |
| |
| if (null != streamInError) throw streamInError; |
| if (0 == len) return 0; //quick. |
| |
| // odd case no data to read -- give back 0 next time -1; |
| if (recordLength == 0 && bytesRead == 0 && !moreChunks) |
| { |
| ++bytesRead; |
| if (ME) finalClose(); |
| return 0; |
| } |
| if (bytesRead >= recordLength && !moreChunks) |
| { |
| dataPadLength -= readPad( dataPadLength ); |
| if (ME) finalClose(); |
| return -1; |
| } |
| |
| |
| do |
| { if (bytesRead >= recordLength && moreChunks) readHeader( true ); |
| bytes2read = (int) Math.min( recordLength - bytesRead, (long)len - totalbytesread ); |
| |
| try |
| { bytes2read = is.read( b, off + totalbytesread, bytes2read ); |
| } |
| catch (IOException e) |
| { |
| streamInError = e; |
| throw e; |
| } |
| |
| if (0 < bytes2read) |
| { |
| totalbytesread += bytes2read; |
| bytesRead += bytes2read; |
| } |
| } |
| while( bytes2read > -1 && totalbytesread < len && (bytesRead < recordLength || moreChunks)); |
| |
| if ( 0 > bytes2read ) |
| { |
| if (moreChunks) |
| { |
| streamInError = new IOException("attach.DimeStreamError0"); |
| throw streamInError; |
| } |
| if (bytesRead < recordLength) |
| { |
| streamInError = new IOException("attach.DimeStreamError1 " + (recordLength - bytesRead)); |
| throw streamInError; |
| } |
| if (!ME) |
| { |
| streamInError = new IOException("attach.DimeStreamError0"); |
| throw streamInError; |
| } |
| //in theory the last chunk of data should also have been padded, but lets be tolerant of that. |
| dataPadLength = 0; |
| } |
| else if (bytesRead >= recordLength) |
| { |
| //get rid of pading. |
| try |
| { dataPadLength -= readPad( dataPadLength ); |
| } |
| catch (IOException e) |
| { |
| //in theory the last chunk of data should also have been padded, but lets be tolerant of that. |
| if (!ME) throw e; |
| else |
| { |
| dataPadLength = 0; |
| streamInError = null; |
| } |
| } |
| } |
| |
| if (bytesRead >= recordLength && ME) finalClose(); |
| |
| return totalbytesread >= 0 ? totalbytesread : -1; |
| } |
| |
| |
| /** |
| * The DIME header is read into local class data fields and are not |
| * passed as part of the stream data. |
| * |
| * @param isChunk |
| * @throws IOException |
| */ |
| protected void readHeader( boolean isChunk ) throws IOException |
| { |
| bytesRead = 0; //How many bytes of the record have been read. |
| |
| if (isChunk) |
| { |
| if (!moreChunks) throw new RuntimeException("attach.DimeStreamError2"); |
| dataPadLength -= readPad(dataPadLength); //Just in case it was left over. |
| } |
| |
| byte[] header = new byte[12]; |
| |
| if (header.length != readFromStream( header) ) |
| { |
| streamInError = new IOException("attach.DimeStreamError3 " + header.length ); |
| throw streamInError; |
| } |
| |
| //VERSION |
| byte version = (byte) ((header[0] >>> 3) & 0x1f); |
| if (version > 1) |
| { |
| streamInError = new IOException("attach.DimeStreamError4 " + version ); |
| throw streamInError; |
| } |
| |
| //B, E, C |
| MB = 0 != (0x4 & header[0]); |
| ME = 0 != (0x2 & header[0]); |
| moreChunks = 0 != (0x1 & header[0]); |
| |
| //TYPE_T |
| if (!isChunk) |
| { |
| switch( ((header[1] >>> 4) & (byte)0x0f) ) { |
| case 0x00: tnf = "UNCHANGED"; break; |
| case 0x01: tnf = "MIME"; break; |
| case 0x02: tnf = "URI"; break; |
| default: tnf = "UNKNOWN"; break; |
| } |
| } |
| |
| //OPTIONS_LENGTH |
| int optionsLength = ((((int) header[2]) << 8) & 0xff00) | ((int) header[3]); |
| |
| //ID_LENGTH |
| int idLength = ((((int) header[4]) << 8) & 0xff00) | ((int) header[5]); |
| |
| //TYPE_LENGTH |
| int typeLength = ((((int) header[6]) << 8) & 0xff00) | ((int) header[7]); |
| |
| //DATA_LENGTH |
| recordLength = ((((long) header[8] ) << 24) & 0xff000000L) | |
| ((((long) header[9] ) << 16) & 0xff0000L ) | |
| ((((long) header[10]) << 8 ) & 0xff00L ) | |
| ((long) header[11] & 0xffL ); |
| |
| //OPTIONS + PADDING |
| if (0 != optionsLength) |
| { |
| byte[] optBytes = new byte[optionsLength]; |
| |
| if (optionsLength != readFromStream( optBytes )) |
| { |
| streamInError = new IOException("attach.DimeStreamError5 " + optionsLength ); |
| throw streamInError; |
| } |
| optBytes = null; // throw it away, don't know anything about options. |
| |
| int pad = (int) ((4L - (optionsLength & 0x3L)) & 0x03L); |
| |
| if (pad != readFromStream( header, 0, pad )) |
| { |
| streamInError = new IOException("attach.DimeStreamError7"); |
| throw streamInError; |
| } |
| } |
| |
| // ID + PADDING |
| if (0 < idLength) |
| { |
| byte[] idBytes = new byte[ idLength]; |
| |
| if (idLength != readFromStream( idBytes )) |
| { |
| streamInError = new IOException("attach.DimeStreamError8"); |
| throw streamInError; |
| } |
| if (idLength != 0 && !isChunk) id = new String(idBytes); |
| |
| int pad = (int) ((4L - (idLength & 0x3L)) & 0x03L); |
| |
| if (pad != readFromStream( header, 0, pad )) |
| { |
| streamInError = new IOException("attach.DimeStreamError9"); |
| throw streamInError; |
| } |
| } |
| |
| //TYPE + PADDING |
| if (0 < typeLength) |
| { |
| byte[] typeBytes = new byte[typeLength]; |
| |
| if (typeLength != readFromStream( typeBytes )) |
| { |
| streamInError = new IOException("attach.DimeStreamError10"); |
| throw streamInError; |
| } |
| if (typeLength != 0 && !isChunk) type = new String(typeBytes); |
| |
| int pad = (int) ((4L - (typeLength & 0x3L)) & 0x03L); |
| |
| if (pad != readFromStream( header, 0, pad )) |
| { |
| streamInError = new IOException("attach.DimeStreamError11"); |
| throw streamInError; |
| } |
| } |
| logger.debug("MB:" + MB + ", ME:" + ME + ", CF:" + moreChunks + |
| "Option length:" + optionsLength + |
| ", ID length:" + idLength + |
| ", typeLength:" + typeLength + ", TYPE_T:" + tnf); |
| logger.debug("id:\"" + id + "\""); |
| logger.debug("type:\"" + type + "\""); |
| logger.debug("recordlength:\"" + recordLength + "\""); |
| |
| dataPadLength = (int) ((4L - (recordLength & 0x3L)) & 0x03L); |
| } |
| |
| |
| /** |
| * Read from the delimited stream. |
| * |
| * @param b is the array to read into. Read as much as possible |
| * into the size of this array. |
| * @return the number of bytes read. -1 if endof stream |
| * @throws IOException if data could not be read from the stream |
| */ |
| public int read( byte[] b ) throws IOException |
| { |
| return read( b, 0, b.length ); |
| } |
| |
| |
| // fixme: this seems a bit inefficient |
| /** |
| * Read from the boundary delimited stream. |
| * |
| * @return the byte read, or -1 if endof stream |
| * @throws IOException if there was an error reading the data |
| */ |
| public int read() throws IOException |
| { |
| byte[] b = new byte[1]; |
| int read = read( b, 0, 1 ); |
| |
| if (read < 0) return -1; // fixme: should we also check for read != 1? |
| return (b[0] & 0xff); // convert byte value to a positive int |
| } |
| |
| |
| /** |
| * Closes the stream. |
| * This will take care of flushing any remaining data to the stream. |
| * Multiple calls to this method will result in the stream being closed once |
| * and then all subsequent calls being ignored. |
| * |
| * @throws IOException if the stream could not be closed |
| */ |
| public void close() throws IOException |
| { |
| synchronized( this ) |
| { |
| if (closed) return; |
| closed = true; //mark it closed. |
| } |
| logger.debug("bStreamClosed " + streamNo); |
| |
| if (bytesRead < recordLength || moreChunks) |
| { |
| //We need get this off the stream. Easy way to flush through the stream; |
| byte[] readrest = new byte[1024 * 16]; |
| int bread = 0; |
| |
| do |
| { bread = _read( readrest, 0, readrest.length ); //should also close the original stream. |
| } |
| while( bread > -1 ); |
| } |
| dataPadLength -= readPad( dataPadLength ); |
| } |
| |
| |
| /** |
| * Skip n bytes of data in the DIME stream, while reading and processing |
| * any headers in the current stream. |
| * |
| * @param n - number of data bytes to skip |
| * @return number of bytes actually skipped |
| * @throws IOException |
| */ |
| public long skip( long n ) throws IOException |
| { |
| long bytesSkipped = 0; |
| long bytes2Read = 0; |
| byte[] dumpbytes = new byte[1024]; |
| |
| while( n > 0 ) |
| { |
| bytes2Read = (n > 1024 ? 1024 : n); |
| bytes2Read = _read( dumpbytes, 0, (int)bytes2Read ); |
| |
| n -= bytes2Read; |
| bytesSkipped += bytes2Read; |
| } |
| |
| return bytesSkipped; |
| } |
| |
| |
| /** |
| * Mark the stream. This is not supported. |
| */ |
| public void mark( int readlimit ) |
| { //do nothing |
| } |
| |
| public void reset() throws IOException |
| { |
| streamInError = new IOException("attach.bounday.mns"); |
| throw streamInError; |
| } |
| |
| public boolean markSupported() |
| { |
| return false; |
| } |
| |
| |
| public synchronized int available() throws IOException |
| { |
| if (null != streamInError) throw streamInError; |
| |
| int chunkAvail = (int) Math.min((long)Integer.MAX_VALUE, recordLength - bytesRead); |
| int streamAvail = 0; |
| |
| try |
| { streamAvail = is.available(); |
| } |
| catch( IOException e ) |
| { |
| streamInError = e; |
| throw e; |
| } |
| |
| if (chunkAvail == 0 && moreChunks && (12 + dataPadLength) <= streamAvail) |
| { |
| dataPadLength -= readPad(dataPadLength); |
| readHeader( true ); |
| return available(); |
| } |
| return Math.min( streamAvail, chunkAvail ); |
| } |
| |
| |
| protected void finalClose() throws IOException |
| { |
| try |
| { theEnd = true; |
| if(null != is) is.close(); |
| } |
| finally |
| { |
| is= null; |
| } |
| } |
| } |
| |
