| // 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.FilterInputStream; |
| import java.io.IOException; |
| import java.io.InputStream; |
| |
| 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 oneButLastByte = (((int)header[2]) << 8) & 0xff00; |
| int lastByte = (int)header[3] & 0xff; |
| int optionsLength = oneButLastByte | lastByte; |
| |
| //ID_LENGTH |
| oneButLastByte = ((((int)header[4]) << 8) & 0xff00); |
| lastByte = ((int)header[5]) & 0xff; |
| int idLength = oneButLastByte | lastByte; |
| |
| //TYPE_LENGTH |
| oneButLastByte = ((((int)header[6]) << 8) & 0xff00); |
| lastByte = ((int)header[7]) & 0xff; |
| int typeLength = oneButLastByte | lastByte; |
| |
| //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; |
| } |
| } |
| } |
