BoundaryDelimitedStream


1   /*
2    * Copyright 2001-2004 The Apache Software Foundation.
3    * 
4    * Licensed under the Apache License, Version 2.0 (the "License");
5    * you may not use this file except in compliance with the License.
6    * You may obtain a copy of the License at
7    * 
8    *      http://www.apache.org/licenses/LICENSE-2.0
9    * 
10   * Unless required by applicable law or agreed to in writing, software
11   * distributed under the License is distributed on an "AS IS" BASIS,
12   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13   * See the License for the specific language governing permissions and
14   * limitations under the License.
15   */
16  package org.apache.axis.attachments;
17  
18  import org.apache.axis.components.logger.LogFactory;
19  import org.apache.axis.utils.Messages;
20  import org.apache.commons.logging.Log;
21  
22  
23  /**
24   * This class takes the input stream and turns it multiple streams.
25   *
26   * @author Rick Rineholt
27   */
28  public class BoundaryDelimitedStream extends java.io.FilterInputStream   {
29  
30      /** The <code>Log</code> that this class should log all events to. */
31      protected static Log log =
32              LogFactory.getLog(BoundaryDelimitedStream.class.getName());
33  
34      protected byte[] boundary = null;
35  
36      /** The boundary length. */
37      int boundaryLen = 0;
38  
39      /** The boundary length plus crlf. */
40      int boundaryBufLen = 0;
41  
42      /** The source input stream. */
43      java.io.InputStream   is = null;
44  
45      /** The stream has been closed. */
46      boolean closed = true;
47  
48      /** eof has been detected. */
49      boolean eos = false;
50  
51      /** There are no more streams left. */
52      boolean theEnd = false;
53  
54      /** Minimum to read at one time. */
55      int readbufsz = 0;
56  
57      /** The buffer we are reading. */
58      byte[] readbuf = null;
59  
60      /** Where we have read so far in the stream. */
61      int readBufPos = 0;
62  
63      /** The number of bytes in array. */
64      int readBufEnd = 0;
65  
66      /** Field BOUNDARY_NOT_FOUND.           */
67      protected static final int BOUNDARY_NOT_FOUND = Integer.MAX_VALUE;
68  
69      // Where in the stream a boundary is located.
70  
71      /** Field boundaryPos.           */
72      int boundaryPos = BOUNDARY_NOT_FOUND;
73  
74      /** The number of streams produced. */
75      static int streamCount = 0;
76  
77      /**
78       * Signal that a new stream has been created.
79       *
80       * @return
81       */
82      protected static synchronized int newStreamNo() {
83  
84          log.debug(Messages.getMessage("streamNo", "" + (streamCount + 1)));
85  
86          return ++streamCount;
87      }
88  
89      /** Field streamNo.           */
90      protected int streamNo = -1;    // Keeps track of stream
91  
92      /** Field isDebugEnabled.           */
93      static boolean isDebugEnabled = false;
94  
95      /**
96       * Gets the next stream. From the previous using the same buffer size to
97       * read.
98       *
99       * @return the boundary delmited stream, null if there are no more streams.
100      * @throws java.io.IOException if there was an error loading the data for
101      *              the next stream
102      */
103     public synchronized BoundaryDelimitedStream getNextStream()  throws java.io.IOException    {
104         return getNextStream(readbufsz);
105     }
106 
107     /**
108      * Gets the next stream. From the previous using  new buffer reading size.
109      *
110      * @param readbufsz
111      * @return the boundary delmited stream, null if there are no more streams.
112      * @throws java.io.IOException if there was an error loading the data for
113      *              the next stream
114      */
115     protected synchronized BoundaryDelimitedStream getNextStream(
116             int readbufsz)  throws java.io.IOException    {
117 
118         BoundaryDelimitedStream ret = null;
119 
120         if (!theEnd) {
121 
122             // Create an new boundary stream  that comes after this one.
123             ret = new BoundaryDelimitedStream(this, readbufsz);
124         }
125 
126         return ret;
127     }
128 
129     /**
130      *  Constructor to create the next stream from the previous one.
131      *
132      * @param prev      the previous stream
133      * @param readbufsz how many bytes to make the read buffer
134      * @throws java.io.IOException if there was a problem reading data from
135      *              <code>prev</code>
136      */
137     protected BoundaryDelimitedStream(BoundaryDelimitedStream prev,
138                                       int readbufsz)
139             throws java.io.IOException  
140     {
141         super(null);
142 
143         streamNo = newStreamNo();
144         boundary = prev.boundary;
145         boundaryLen = prev.boundaryLen;
146         boundaryBufLen = prev.boundaryBufLen;
147         skip = prev.skip;
148         is = prev.is;
149         closed = false;    // The new one is not closed.
150         eos = false;    // Its not at th EOS.
151         readbufsz = prev.readbufsz;
152         readbuf = prev.readbuf;
153 
154         // Move past the old boundary.
155         readBufPos = prev.readBufPos + boundaryBufLen;
156         readBufEnd = prev.readBufEnd;
157 
158         // find the new boundary.
159         boundaryPos = boundaryPosition(readbuf, readBufPos, readBufEnd);
160         prev.theEnd = theEnd;      // The stream.
161     }
162 
163     /**
164      * Create a new boundary stream.
165      *
166      * @param is
167      * @param boundary is the boundary that separates the individual streams.
168      * @param readbufsz lets you have some control over the amount of buffering.
169      *   by buffering you can some effiency in searching.
170      *
171      * @throws org.apache.axis.AxisFault
172      */
173     BoundaryDelimitedStream(
174             java.io.InputStream   is, byte[] boundary, int readbufsz)
175             throws org.apache.axis.AxisFault {
176 
177         // super (is);
178         super(null);    // we handle everything so this is not necessary, don't won't to hang on to a reference.
179 
180         isDebugEnabled = log.isDebugEnabled();
181         streamNo = newStreamNo();
182         closed = false;
183         this.is = is;
184 
185         // Copy the boundary array to make certain it is never altered.
186         this.boundary = new byte[boundary.length];
187 
188         System.arraycopy(boundary, 0, this.boundary, 0, boundary.length);
189 
190         this.boundaryLen = this.boundary.length;
191         this.boundaryBufLen = boundaryLen + 2;
192 
193         // allways leave room for at least a 2x boundary
194         // Most mime boundaries are 40 bytes or so.
195         this.readbufsz = Math.max((boundaryBufLen) * 2, readbufsz);
196     }
197 
198     private final int readFromStream(final byte[] b)
199             throws java.io.IOException   {
200         return readFromStream(b, 0, b.length);
201     }
202 
203     private final int readFromStream(
204             final byte[] b, final int start, final int length)
205             throws java.io.IOException   {
206 
207         int minRead = Math.max(boundaryBufLen * 2, length);
208 
209         minRead = Math.min(minRead, length - start);
210 
211         int br = 0;
212         int brTotal = 0;
213 
214         do {
215             br = is.read(b, brTotal + start, length - brTotal);
216 
217             if (br > 0) {
218                 brTotal += br;
219             }
220         } while ((br > -1) && (brTotal < minRead));
221 
222         return (brTotal != 0)
223                 ? brTotal
224                 : br;
225     }
226 
227     /**
228      * Read from the boundary delimited stream.
229      * @param b is the array to read into.
230      * @param off is the offset
231      * @param len
232      * @return the number of bytes read. -1 if endof stream.
233      *
234      * @throws java.io.IOException
235      */
236     public synchronized int read(byte[] b, final int off, final int len)
237             throws java.io.IOException   {
238 
239         if (closed) {
240             throw new java.io.IOException  (Messages.getMessage("streamClosed"));
241         }
242 
243         if (eos) {
244             return -1;
245         }
246 
247         if (readbuf == null) {    // Allocate the buffer.
248             readbuf = new byte[Math.max(len, readbufsz)];
249             readBufEnd = readFromStream(readbuf);
250 
251             if (readBufEnd < 0) {
252                 readbuf = null;
253                 closed = true;
254                 finalClose();
255 
256                 throw new java.io.IOException  (
257                         Messages.getMessage("eosBeforeMarker"));
258             }
259 
260             readBufPos = 0;
261 
262             // Finds the boundary pos.
263             boundaryPos = boundaryPosition(readbuf, 0, readBufEnd);
264         }
265 
266         int bwritten = 0;    // Number of bytes written.
267 
268         // read and copy bytes in.
269         do {                                // Always allow to have a boundary length left in the buffer.
270             int bcopy = Math.min(readBufEnd - readBufPos - boundaryBufLen,
271                     len - bwritten);
272 
273             // never go past the boundary.
274             bcopy = Math.min(bcopy, boundaryPos - readBufPos);
275 
276             if (bcopy > 0) {
277                 System.arraycopy(readbuf, readBufPos, b, off + bwritten, bcopy);
278 
279                 bwritten += bcopy;
280                 readBufPos += bcopy;
281             }
282 
283             if (readBufPos == boundaryPos) {
284                 eos = true;                 // hit the boundary so it the end of the stream.
285 
286                 log.debug(Messages.getMessage("atEOS", "" + streamNo));
287             } else if (bwritten < len) {    // need to get more data.
288                 byte[] dstbuf = readbuf;
289 
290                 if (readbuf.length < len) {
291                     dstbuf = new byte[len];
292                 }
293 
294                 int movecnt = readBufEnd - readBufPos;
295 
296                 // copy what was left over.
297                 System.arraycopy(readbuf, readBufPos, dstbuf, 0, movecnt);
298 
299                 // Read in the new data.
300                 int readcnt = readFromStream(dstbuf, movecnt,
301                         dstbuf.length - movecnt);
302 
303                 if (readcnt < 0) {
304                     readbuf = null;
305                     closed = true;
306                     finalClose();
307 
308                     throw new java.io.IOException  (
309                             Messages.getMessage("eosBeforeMarker"));
310                 }
311 
312                 readBufEnd = readcnt + movecnt;
313                 readbuf = dstbuf;
314                 readBufPos = 0;             // start at the begining.
315 
316                 // just move the boundary by what we moved
317                 if (BOUNDARY_NOT_FOUND != boundaryPos) {
318                     boundaryPos -= movecnt;
319                 } else {
320                     boundaryPos = boundaryPosition(
321                             readbuf, readBufPos,
322                             readBufEnd);        // See if the boundary is now there.
323                 }
324             }
325         }
326 
327                 // read till we get the amount or the stream is finished.
328         while (!eos && (bwritten < len));
329 
330         if (log.isDebugEnabled()) {
331             if (bwritten > 0) {
332                 byte tb[] = new byte[bwritten];
333 
334                 System.arraycopy(b, off, tb, 0, bwritten);
335                 log.debug(Messages.getMessage("readBStream",
336                         new String  []{"" + bwritten,
337                                      "" + streamNo,
338                                      new String  (tb)}));
339             }
340         }
341 
342         if (eos && theEnd) {
343             readbuf = null;    // dealloc even in Java.
344         }
345 
346         return bwritten;
347     }
348 
349     /**
350      * Read from the boundary delimited stream.
351      * @param b is the array to read into. Read as much as possible
352      *   into the size of this array.
353      * @return the number of bytes read. -1 if endof stream.
354      *
355      * @throws java.io.IOException
356      */
357     public int read(byte[] b) throws java.io.IOException   {
358         return read(b, 0, b.length);
359     }
360 
361     /**
362      * Read from the boundary delimited stream.
363      * @return The byte read, or -1 if endof stream.
364      *
365      * @throws java.io.IOException
366      */
367     public int read() throws java.io.IOException   {
368 
369         byte[] b = new byte[1];    // quick and dirty. //for now
370         int read = read(b);
371 
372         if (read < 0) {
373             return -1;
374         } else {
375             return b[0]&0xff;
376         }
377     }
378 
379     /**
380      * Closes the stream.
381      *
382      * @throws java.io.IOException
383      */
384     public synchronized void close() throws java.io.IOException   {
385 
386         if (closed) {
387             return;
388         }
389 
390         log.debug(Messages.getMessage("bStreamClosed", "" + streamNo));
391 
392         closed = true;    // mark it closed.
393 
394         if (!eos) {    // We need get this off the stream.
395 
396             // Easy way to flush through the stream;
397             byte[] readrest = new byte[1024 * 16];
398             int bread = 0;
399 
400             do {
401                 bread = read(readrest);
402             } while (bread > -1);
403         }
404     }
405 
406     /**
407      * mark the stream.
408      * This is not supported.
409      *
410      * @param readlimit
411      */
412     public void mark(int readlimit) {
413 
414         // do nothing
415     }
416 
417     /**
418      * reset the stream.
419      * This is not supported.
420      *
421      * @throws java.io.IOException
422      */
423     public void reset() throws java.io.IOException   {
424         throw new java.io.IOException  (
425                 Messages.getMessage("attach.bounday.mns"));
426     }
427 
428     /**
429      * markSupported
430      * return false;
431      *
432      * @return
433      */
434     public boolean markSupported() {
435         return false;
436     }
437 
438     public int available() throws java.io.IOException   {
439 
440         int bcopy = readBufEnd - readBufPos - boundaryBufLen;
441 
442         // never go past the boundary.
443         bcopy = Math.min(bcopy, boundaryPos - readBufPos);
444 
445         return Math.max(0, bcopy);
446     }
447 
448     /**
449      * Read from the boundary delimited stream.
450      *
451      * @param searchbuf buffer to read from
452      * @param start     starting index
453      * @param end       ending index
454      * @return The position of the boundary. Detects the end of the source stream.
455      * @throws java.io.IOException if there was an error manipulating the
456      *              underlying stream
457      */
458     protected int boundaryPosition(byte[] searchbuf, int start, int end) throws java.io.IOException    {
459 
460         int foundAt = boundarySearch(searchbuf, start, end);
461 
462         // First find the boundary marker
463         if (BOUNDARY_NOT_FOUND != foundAt) {    // Something was found.
464             if (foundAt + boundaryLen + 2 > end) {
465                 foundAt = BOUNDARY_NOT_FOUND;
466             } else {
467 
468                 // If the marker has a "--" at the end then this is the last boundary.
469                 if ((searchbuf[foundAt + boundaryLen] == '-')
470                         && (searchbuf[foundAt + boundaryLen + 1] == '-')) {
471                     finalClose();
472                 } else if ((searchbuf[foundAt + boundaryLen] != 13)
473                         || (searchbuf[foundAt + boundaryLen + 1] != 10)) {
474 
475                     // If there really was no crlf at then end then this is not a boundary.
476                     foundAt = BOUNDARY_NOT_FOUND;
477                 }
478             }
479         }
480 
481         return foundAt;
482     }
483 
484     /* The below uses a standard textbook Boyer-Moore pattern search. */
485 
486     private int[] skip = null;
487 
488     private int boundarySearch(final byte[] text, final int start,
489                                final int end) {
490 
491         // log.debug(">>>>" + start + "," + end);
492         int i = 0, j = 0, k = 0;
493 
494         if (boundaryLen > (end - start)) {
495             return BOUNDARY_NOT_FOUND;
496         }
497 
498         if (null == skip) {
499             skip = new int[256];
500 
501             java.util.Arrays.fill(skip, boundaryLen);
502 
503             for (k = 0; k < boundaryLen - 1; k++) {
504                 skip[boundary[k]] = boundaryLen - k - 1;
505             }
506         }
507 
508         for (k = start + boundaryLen - 1; k < end;
509              k += skip[text[k] & (0xff)]) {
510 
511             // log.debug(">>>>" + k);
512             // printarry(text, k-boundaryLen+1, end);
513             try {
514                 for (j = boundaryLen - 1, i = k;
515                      (j >= 0) && (text[i] == boundary[j]); j--) {
516                     i--;
517                 }
518             } catch (ArrayIndexOutOfBoundsException   e) {
519                 StringBuffer   sb = new StringBuffer  ();
520                 sb.append(
521                         ">>>"
522                         + e);    // rr temporary till a boundary issue is resolved.
523                 sb.append("start=" + start);
524                 sb.append("k=" + k);
525                 sb.append("text.length=" + text.length);
526                 sb.append("i=" + i);
527                 sb.append("boundary.length=" + boundary.length);
528                 sb.append("j=" + j);
529                 sb.append("end=" + end);
530                 log.warn(Messages.getMessage("exception01",sb.toString()));
531                 throw e;
532             }
533 
534             if (j == (-1)) {
535                 return i + 1;
536             }
537         }
538 
539         // log.debug(">>>> not found" );
540         return BOUNDARY_NOT_FOUND;
541     }
542 
543     /**
544      * Close the underlying stream and remove all references to it.
545      *
546      * @throws java.io.IOException if the stream could not be closed
547      */
548     protected void finalClose() throws java.io.IOException   {
549       if(theEnd) return;
550       theEnd= true;
551       is.close();
552       is= null;
553     }
554 
555     /**
556      * Method printarry
557      *
558      * @param b
559      * @param start
560      * @param end
561      */
562     public static void printarry(byte[] b, int start, int end) {
563 
564         if (log.isDebugEnabled()) {
565             byte tb[] = new byte[end - start];
566 
567             System.arraycopy(b, start, tb, 0, end - start);
568             log.debug("\"" + new String  (tb) + "\"");
569         }
570     }
571 }
572
A to Z: JavaDoc & Examples Daily Java News & Articles Open Source Projects Open Source Codes Free Computer Books Remove Frame
Popular Tags