ConcatReader


1   /*
2    * Copyright (C) 2004 Stephen Ostermiller
3    * http://ostermiller.org/contact.pl?regarding=Java+Utilities
4    *
5    * This program is free software; you can redistribute it and/or modify
6    * it under the terms of the GNU General Public License as published by
7    * the Free Software Foundation; either version 2 of the License, or
8    * (at your option) any later version.
9    *
10   * This program is distributed in the hope that it will be useful,
11   * but WITHOUT ANY WARRANTY; without even the implied warranty of
12   * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
13   * GNU General Public License for more details.
14   *
15   * See COPYING.TXT for details.
16   */
17  package com.Ostermiller.util;
18  
19  import java.io.*;
20  import java.util.*;
21  
22  /**
23   * A reader which reads sequentially from multiple sources.
24   * More information about this class is available from <a target="_top" HREF=
25   * "http://ostermiller.org/utils/">ostermiller.org</a>.
26   *
27   * @author Stephen Ostermiller http://ostermiller.org/contact.pl?regarding=Java+Utilities
28   * @since ostermillerutils 1.04.00
29   */
30  public class ConcatReader extends Reader {
31  
32      /**
33       * Current index to readerQueue
34       *
35       * @since ostermillerutils 1.04.01
36       */
37      private int readerQueueIndex = 0;
38  
39      /**
40       * Queue of readers that have yet to be read from.
41       *
42       * @since ostermillerutils 1.04.01
43       */
44      private ArrayList<Reader> readerQueue = new ArrayList<Reader>();
45  
46      /**
47       * A cache of the current reader from the readerQueue
48       * to avoid unneeded access to the queue which must
49       * be synchronized.
50       *
51       * @since ostermillerutils 1.04.01
52       */
53      private Reader currentReader = null;
54  
55      /**
56       * true iff the client may add more readers.
57       *
58       * @since ostermillerutils 1.04.01
59       */
60      private boolean doneAddingReaders = false;
61  
62      /**
63       * Causes the addReader method to throw IllegalStateException
64       * and read() methods to return -1 (end of stream)
65       * when there is no more available data.
66       * <p>
67       * Calling this method when this class is no longer accepting
68       * more readers has no effect.
69       *
70       * @since ostermillerutils 1.04.01
71       */
72      public void lastReaderAdded(){
73          doneAddingReaders = true;
74      }
75  
76      /**
77       * Add the given reader to the queue of readers from which to
78       * concatenate data.
79       *
80       * @param in Reader to add to the concatenation.
81       * @throws IllegalStateException if more readers can't be added because lastReaderAdded() has been called, close() has been called, or a constructor with reader parameters was used.
82       *
83       * @since ostermillerutils 1.04.01
84       */
85      public void addReader(Reader in){
86          synchronized(readerQueue){
87              if (in == null) throw new NullPointerException  ();
88              if (closed) throw new IllegalStateException  ("ConcatReader has been closed");
89              if (doneAddingReaders) throw new IllegalStateException  ("Cannot add more readers - the last reader has already been added.");
90              readerQueue.add(in);
91          }
92      }
93  
94      /**
95       * Add the given reader to the queue of readers from which to
96       * concatenate data.
97       *
98       * @param in Reader to add to the concatenation.
99       * @throws IllegalStateException if more readers can't be added because lastReaderAdded() has been called, close() has been called, or a constructor with reader parameters was used.
100      * @throws NullPointerException the array of readers, or any of the contents is null.
101      *
102      * @since ostermillerutils 1.04.01
103      */
104     public void addReaders(Reader[] in){
105         for (int i=0; i<in.length; i++){
106             addReader(in[i]);
107         }
108     }
109 
110     /**
111      * Gets the current reader, looking at the next
112      * one in the list if the current one is null.
113      *
114      * @since ostermillerutils 1.04.01
115      */
116     private Reader getCurrentReader(){
117         if (currentReader == null && readerQueueIndex < readerQueue.size()){
118             synchronized(readerQueue){
119                 // reader queue index is advanced only by the nextReader()
120                 // method.  Don't do it here.
121                 currentReader = (Reader)readerQueue.get(readerQueueIndex);
122             }
123         }
124         return currentReader;
125     }
126 
127     /**
128      * Indicate that we are done with the current reader and we should
129      * advance to the next reader.
130      *
131      * @since ostermillerutils 1.04.01
132      */
133     private void advanceToNextReader(){
134         currentReader = null;
135         readerQueueIndex++;
136     }
137 
138     /**
139      * True iff this the close() method has been called on this stream.
140      *
141      * @since ostermillerutils 1.04.00
142      */
143     private boolean closed = false;
144 
145     /**
146      * Create a new reader that can dynamically accept new sources.
147      * <p>
148      * New sources should be added using the addReader() method.
149      * When all sources have been added the lastReaderAdded() should
150      * be called so that read methods can return -1 (end of stream).
151      * <p>
152      * Adding new sources may by interleaved with read calls.
153      *
154      * @since ostermillerutils 1.04.01
155      */
156     public ConcatReader(){
157     }
158 
159     /**
160      * Create a new reader with one source.
161      * <p>
162      * When using this constructor, more readers cannot
163      * be added later, and calling addReader() will
164      * throw an illegal state Exception.
165      *
166      * @param in reader to use as a source.
167      *
168      * @throws NullPointerException if in is null
169      *
170      * @since ostermillerutils 1.04.00
171      */
172     public ConcatReader(Reader in){
173         addReader(in);
174         lastReaderAdded();
175     }
176 
177     /**
178      * Create a new reader with two sources.
179      * <p>
180      * When using this constructor, more readers cannot
181      * be added later, and calling addReader() will
182      * throw an illegal state Exception.
183      *
184      * @param in1 first reader to use as a source.
185      * @param in2 second reader to use as a source.
186      *
187      * @throws NullPointerException if either source is null.
188      *
189      * @since ostermillerutils 1.04.00
190      */
191     public ConcatReader(Reader in1, Reader in2){
192         addReader(in1);
193         addReader(in2);
194         lastReaderAdded();
195     }
196 
197     /**
198      * Create a new reader with an arbitrary number of sources.
199      * <p>
200      * When using this constructor, more readers cannot
201      * be added later, and calling addReader() will
202      * throw an illegal state Exception.
203      *
204      * @param in readers to use as a sources.
205      *
206      * @throws NullPointerException if the input array on any element is null.
207      *
208      * @since ostermillerutils 1.04.00
209      */
210     public ConcatReader(Reader[] in){
211         addReaders(in);
212         lastReaderAdded();
213     }
214 
215     /**
216      * Read a single character. This method will block until a
217      * character is available, an I/O error occurs, or the end of all underlying
218      * streams are reached.
219      * <p>
220      * If this class in not done accepting readers and the end of the last known
221      * stream is reached, this method will block forever unless another thread
222      * adds a reader or interrupts.
223      *
224      * @return The character read, as an integer in the range 0 to 65535 (0x00-0xffff),
225      *    or -1 if the end of the stream has been reached
226      *
227      * @throws IOException - If an I/O error occurs
228      *
229      * @since ostermillerutils 1.04.00
230      */
231     public int read() throws IOException {
232         if (closed) throw new IOException("Reader closed");
233         int r = -1;
234         while (r == -1){
235             Reader in = getCurrentReader();
236             if (in == null){
237                 if (doneAddingReaders) return -1;
238                 try {
239                     Thread.sleep(100);
240                 } catch (InterruptedException   iox){
241                     throw new IOException("Interrupted");
242                 }
243             } else {
244                 r = in.read();
245                 if (r == -1) advanceToNextReader();
246             }
247         }
248         return r;
249     }
250 
251     /**
252      * Read characters into an array. This method will block until some input is available, an
253      * I/O error occurs, or the end of all underlying
254      * streams are reached.
255      * <p>
256      * If this class in not done accepting readers and the end of the last known
257      * stream is reached, this method will block forever unless another thread
258      * adds a reader or interrupts.
259      *
260      * @param cbuf - Destination buffer
261      * @return The number of characters read, or -1 if the end of the stream has been reached
262      *
263      * @throws IOException - If an I/O error occurs
264      * @throws NullPointerException - If cbuf is null.
265      *
266      * @since ostermillerutils 1.04.00
267      */
268     public int read(char[] cbuf) throws IOException {
269         return read(cbuf, 0, cbuf.length);
270     }
271 
272     /**
273      * Read characters into a portion of an array. This method will block until
274      * some input is available, an I/O error occurs, or the end of all underlying
275      * streams are reached.
276      * <p>
277      * If this class in not done accepting readers and the end of the last known
278      * stream is reached, this method will block forever unless another thread
279      * adds a reader or interrupts.
280      *
281      * @param cbuf Destination buffer
282      * @param off Offset at which to start storing characters
283      * @param len Maximum number of characters to read
284      * @return The number of characters read, or -1 if the end of the stream has been reached
285      *
286      * @throws IOException - If an I/O error occurs
287      * @throws NullPointerException - If cbuf is null.
288      * @throws IndexOutOfBoundsException - if len or offset are not possible.
289      *
290      * @since ostermillerutils 1.04.00
291      */
292     public int read(char[] cbuf, int off, int len) throws IOException {
293         if (off < 0 || len < 0 || off + len > cbuf.length) throw new IndexOutOfBoundsException  ();
294         if (closed) throw new IOException("Reader closed");
295         int r = -1;
296         while (r == -1){
297             Reader in = getCurrentReader();
298             if (in == null){
299                 if (doneAddingReaders) return -1;
300                 try {
301                     Thread.sleep(100);
302                 } catch (InterruptedException   iox){
303                     throw new IOException("Interrupted");
304                 }
305             } else {
306                 r = in.read(cbuf, off, len);
307                 if (r == -1) advanceToNextReader();
308             }
309         }
310         return r;
311     }
312 
313     /**
314      * Skip characters. This method will block until some characters are
315      * available, an I/O error occurs, or the end of the stream is reached.
316      * <p>
317      * If this class in not done accepting readers and the end of the last known
318      * stream is reached, this method will block forever unless another thread
319      * adds a reader or interrupts.
320      *
321      * @param n the number of characters to skip
322      * @return The number of characters actually skipped
323      *
324      * @throws IllegalArgumentException If n is negative.
325      * @throws IOException If an I/O error occurs
326      *
327      * @since ostermillerutils 1.04.00
328      */
329     public long skip(long n) throws IOException {
330         if (closed) throw new IOException("Reader closed");
331         if (n <= 0) return 0;
332         long s = -1;
333         while (s <= 0){
334             Reader in = getCurrentReader();
335             if (in == null){
336                 if (doneAddingReaders) return 0;
337                 try {
338                     Thread.sleep(100);
339                 } catch (InterruptedException   iox){
340                     throw new IOException("Interrupted");
341                 }
342             } else {
343                 s = in.skip(n);
344                 // When nothing was skipped it is a bit of a puzzle.
345                 // The most common cause is that the end of the underlying
346                 // stream was reached.  In which case calling skip on it
347                 // will always return zero.  If somebody were calling skip
348                 // until it skipped everything they needed, there would
349                 // be an infinite loop if we were to return zero here.
350                 // If we get zero, let us try to read one character so
351                 // we can see if we are at the end of the stream.  If so,
352                 // we will move to the next.
353                 if (s <= 0) {
354                     // read() will advance to the next stream for us, so don't do it again
355                     s = ((read()==-1)?-1:1);
356                 }
357             }
358 
359         }
360         return s;
361     }
362 
363     /**
364      * Tell whether this stream is ready to be read.
365      *
366      * @return True if the next read() is guaranteed not to block for input,
367      *    false otherwise. Note that returning false does not guarantee that the next
368      *    read will block.
369      *
370      * @throws IOException If an I/O error occurs
371      *
372      * @since ostermillerutils 1.04.00
373      */
374     public boolean ready() throws IOException {
375         if (closed) throw new IOException("Reader closed");
376         Reader in = getCurrentReader();
377         if (in == null) return false;
378         return in.ready();
379     }
380 
381     /**
382      * Close the stream and any underlying streams.
383      * Once a stream has been closed, further read(), ready(), mark(), or reset()
384      * invocations will throw an IOException. Closing a previously-closed stream,
385      * however, has no effect.
386      *
387      * @throws IOException If an I/O error occurs
388      *
389      * @since ostermillerutils 1.04.00
390      */
391     public void close() throws IOException {
392         if (closed) return;
393         for (Iterator i=readerQueue.iterator(); i.hasNext();){
394             ((Reader)i.next()).close();
395         }
396         closed = true;
397     }
398 
399     /**
400      * Mark not supported.
401      *
402      * @throws IOException because mark is not supported.
403      *
404      * @since ostermillerutils 1.04.00
405      */
406     public void mark(int readlimit) throws IOException {
407         throw new IOException("Mark not supported");
408     }
409 
410     /**
411      * Reset not supported.
412      *
413      * @throws IOException because reset is not supported.
414      *
415      * @since ostermillerutils 1.04.00
416      */
417     public void reset() throws IOException {
418         throw new IOException("Reset not supported");
419     }
420 
421     /**
422      * Mark not supported.
423      *
424      * @return false
425      *
426      * @since ostermillerutils 1.04.00
427      */
428     public boolean markSupported(){
429         return false;
430     }
431 }
432
A to Z: JavaDoc & Examples Daily Java News & Articles Open Source Projects Open Source Codes Free Computer Books Remove Frame
Popular Tags