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 / / | +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+* This implementation of input stream does not support marking operations. * * @author Rick Rineholt */ public class DimeDelimitedInputStream extends java.io.FilterInputStream { protected static Log log = LogFactory.getLog(DimeDelimitedInputStream.class.getName()); java.io.InputStream is = null; //The source input stream. volatile 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; //First part of the stream. MUST be SOAP. boolean ME = false; //Last part of stream. DimeTypeNameFormat tnf = null; String type = null; String id = null; long recordLength = 0L; //length of the record. long bytesRead = 0; //How many bytes of the record have been read. int dataPadLength = 0; //How many pad bytes there are. private static byte[] trash = new byte[4]; protected int streamNo = 0; protected IOException streamInError = null; protected static int streamCount = 0; //number of streams produced. protected static synchronized int newStreamNo() { log.debug(Messages.getMessage("streamNo", "" + (streamCount + 1))); return ++streamCount; } static boolean isDebugEnabled = false; /** * Gets the next stream. From the previous using new buffer reading size. * * @return the dime delmited stream, null if there are no more streams * @throws IOException if there was an error loading the data for the next * stream */ synchronized DimeDelimitedInputStream getNextStream() throws IOException { if (null != streamInError) throw streamInError; if (theEnd) return null; if (bytesRead < recordLength || moreChunks) //Stream must be read in succession throw new RuntimeException(Messages.getMessage( "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
InputStream to wrap
* @throws IOException if anything goes wrong
*/
DimeDelimitedInputStream(java.io.InputStream is) throws IOException {
super(null); //we handle everything so this is not necessary, don't won't to hang on to a reference.
isDebugEnabled = log.isDebugEnabled();
streamNo = newStreamNo();
closed = false;
this.is = is;
readHeader(false);
}
private final int readPad(final int size) throws IOException {
if (0 == size) return 0;
int read = readFromStream(trash, 0, size);
if (size != read) {
streamInError = new IOException(Messages.getMessage(
"attach.dimeNotPaddedCorrectly"));
throw streamInError;
}
return read;
}
private final int readFromStream(final byte[] b) throws IOException {
return readFromStream(b, 0, b.length);
}
private final int readFromStream(final byte[] b,
final int start, final int length)
throws IOException {
if (length == 0) return 0;
int br = 0;
int brTotal = 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 DimeTypeNameFormat 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, final int off,
final int len) throws IOException {
if (closed) {
dataPadLength -= readPad(dataPadLength);
throw new IOException(Messages.getMessage("streamClosed"));
}
return _read(b, off, len);
}
protected int _read(byte[] b, final int off, final int len)
throws IOException {
if (len < 0) throw new IllegalArgumentException
(Messages.getMessage("attach.readLengthError",
"" + len));
if (off < 0) throw new IllegalArgumentException
(Messages.getMessage("attach.readOffsetError",
"" + off));
if (b == null) throw new IllegalArgumentException
(Messages.getMessage("attach.readArrayNullError"));
if (b.length < off + len) throw new IllegalArgumentException
(Messages.getMessage("attach.readArraySizeError",
"" + b.length, "" + len, "" + off));
if (null != streamInError) throw streamInError;
if (0 == len) return 0; //quick.
if(recordLength == 0 && bytesRead == 0 && !moreChunks){
++bytesRead; //odd case no data to read -- give back 0 next time -1;
if(ME){
finalClose();
}
return 0;
}
if (bytesRead >= recordLength && !moreChunks) {
dataPadLength -= readPad(dataPadLength);
if(ME){
finalClose();
}
return -1;
}
int totalbytesread = 0;
int bytes2read = 0;
do {
if (bytesRead >= recordLength && moreChunks)
readHeader(true);
bytes2read = (int) Math.min(recordLength - bytesRead,
(long) len - totalbytesread);
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(Messages.getMessage(
"attach.DimeStreamError0"));
throw streamInError;
}
if (bytesRead < recordLength) {
streamInError = new IOException(Messages.getMessage
("attach.DimeStreamError1",
"" + (recordLength - bytesRead)));
throw streamInError;
}
if (!ME) {
streamInError = new IOException(Messages.getMessage(
"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;
}
void readHeader(boolean isChunk) throws IOException {
bytesRead = 0; //How many bytes of the record have been read.
if (isChunk) {
if (!moreChunks) throw new RuntimeException(
Messages.getMessage("attach.DimeStreamError2"));
dataPadLength -= readPad(dataPadLength); //Just incase it was left over.
}
byte[] header = new byte[12];
if (header.length != readFromStream(header)) {
streamInError = new IOException(Messages.getMessage(
"attach.DimeStreamError3",
"" + header.length));
throw streamInError;
}
//VERSION
byte version = (byte) ((header[0] >>> 3) & 0x1f);
if (version > DimeMultiPart.CURRENT_VERSION) {
streamInError = new IOException(Messages.getMessage("attach.DimeStreamError4",
"" + version,
"" + DimeMultiPart.CURRENT_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)
tnf = DimeTypeNameFormat.parseByte((byte) ((header[1] >>> 4) & (byte) 0xf));
//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(Messages.getMessage(
"attach.DimeStreamError5",
"" + optionsLength));
throw streamInError;
}
optBytes = null; //Yup throw it away, don't know anything about options.
int pad = DimeBodyPart.dimePadding(optionsLength);
if (pad != readFromStream(header, 0, pad)) {
streamInError = new IOException(
Messages.getMessage("attach.DimeStreamError7"));
throw streamInError;
}
}
// ID + PADDING
if (0 < idLength) {
byte[] idBytes = new byte[ idLength];
if (idLength != readFromStream(idBytes)) {
streamInError = new IOException(
Messages.getMessage("attach.DimeStreamError8"));
throw streamInError;
}
if (idLength != 0 && !isChunk) {
id = new String(idBytes);
}
int pad = DimeBodyPart.dimePadding(idLength);
if (pad != readFromStream(header, 0, pad)) {
streamInError = new IOException(Messages.getMessage(
"attach.DimeStreamError9"));
throw streamInError;
}
}
//TYPE + PADDING
if (0 < typeLength) {
byte[] typeBytes = new byte[typeLength];
if (typeLength != readFromStream(typeBytes)) {
streamInError = new IOException(Messages.getMessage(
"attach.DimeStreamError10"));
throw streamInError;
}
if (typeLength != 0 && !isChunk) {
type = new String(typeBytes);
}
int pad = DimeBodyPart.dimePadding(typeLength);
if (pad != readFromStream(header, 0, pad)) {
streamInError = new IOException(Messages.getMessage(
"attach.DimeStreamError11"));
throw streamInError;
}
}
log.debug("MB:" + MB + ", ME:" + ME + ", CF:" + moreChunks +
"Option length:" + optionsLength +
", ID length:" + idLength +
", typeLength:" + typeLength + ", TYPE_T:" + tnf);
log.debug("id:\"" + id + "\"");
log.debug("type:\"" + type + "\"");
log.debug("recordlength:\"" + recordLength + "\"");
dataPadLength = DimeBodyPart.dimePadding(recordLength);
}
/**
* 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 strea. *
* 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. } log.debug(Messages.getMessage("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 orginal stream. } while (bread > -1); } dataPadLength -= readPad(dataPadLength); } // fixme: if mark is not supported, do we throw an exception here? /** * Mark the stream. * This is not supported. */ public void mark(int readlimit) {//do nothing } public void reset() throws IOException { streamInError = new IOException(Messages.getMessage( "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; } } }