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