View Javadoc
1   /*
2    * Copyright (c) 1994, 2013, Oracle and/or its affiliates. All rights reserved.
3    * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
4    *
5    * This code is free software; you can redistribute it and/or modify it
6    * under the terms of the GNU General Public License version 2 only, as
7    * published by the Free Software Foundation.  Oracle designates this
8    * particular file as subject to the "Classpath" exception as provided
9    * by Oracle in the LICENSE file that accompanied this code.
10   *
11   * This code is distributed in the hope that it will be useful, but WITHOUT
12   * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
13   * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
14   * version 2 for more details (a copy is included in the LICENSE file that
15   * accompanied this code).
16   *
17   * You should have received a copy of the GNU General Public License version
18   * 2 along with this work; if not, write to the Free Software Foundation,
19   * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
20   *
21   * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
22   * or visit www.oracle.com if you need additional information or have any
23   * questions.
24   */
25  
26  package java.io;
27  import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
28  
29  /**
30   * A <code>BufferedInputStream</code> adds
31   * functionality to another input stream-namely,
32   * the ability to buffer the input and to
33   * support the <code>mark</code> and <code>reset</code>
34   * methods. When  the <code>BufferedInputStream</code>
35   * is created, an internal buffer array is
36   * created. As bytes  from the stream are read
37   * or skipped, the internal buffer is refilled
38   * as necessary  from the contained input stream,
39   * many bytes at a time. The <code>mark</code>
40   * operation  remembers a point in the input
41   * stream and the <code>reset</code> operation
42   * causes all the  bytes read since the most
43   * recent <code>mark</code> operation to be
44   * reread before new bytes are  taken from
45   * the contained input stream.
46   *
47   * @author  Arthur van Hoff
48   * @since   JDK1.0
49   */
50  public
51  class BufferedInputStream extends FilterInputStream {
52  
53      private static int DEFAULT_BUFFER_SIZE = 8192;
54  
55      /**
56       * The maximum size of array to allocate.
57       * Some VMs reserve some header words in an array.
58       * Attempts to allocate larger arrays may result in
59       * OutOfMemoryError: Requested array size exceeds VM limit
60       */
61      private static int MAX_BUFFER_SIZE = Integer.MAX_VALUE - 8;
62  
63      /**
64       * The internal buffer array where the data is stored. When necessary,
65       * it may be replaced by another array of
66       * a different size.
67       */
68      protected volatile byte buf[];
69  
70      /**
71       * Atomic updater to provide compareAndSet for buf. This is
72       * necessary because closes can be asynchronous. We use nullness
73       * of buf[] as primary indicator that this stream is closed. (The
74       * "in" field is also nulled out on close.)
75       */
76      private static final
77          AtomicReferenceFieldUpdater<BufferedInputStream, byte[]> bufUpdater =
78          AtomicReferenceFieldUpdater.newUpdater
79          (BufferedInputStream.class,  byte[].class, "buf");
80  
81      /**
82       * The index one greater than the index of the last valid byte in
83       * the buffer.
84       * This value is always
85       * in the range <code>0</code> through <code>buf.length</code>;
86       * elements <code>buf[0]</code>  through <code>buf[count-1]
87       * </code>contain buffered input data obtained
88       * from the underlying  input stream.
89       */
90      protected int count;
91  
92      /**
93       * The current position in the buffer. This is the index of the next
94       * character to be read from the <code>buf</code> array.
95       * <p>
96       * This value is always in the range <code>0</code>
97       * through <code>count</code>. If it is less
98       * than <code>count</code>, then  <code>buf[pos]</code>
99       * is the next byte to be supplied as input;
100      * if it is equal to <code>count</code>, then
101      * the  next <code>read</code> or <code>skip</code>
102      * operation will require more bytes to be
103      * read from the contained  input stream.
104      *
105      * @see     java.io.BufferedInputStream#buf
106      */
107     protected int pos;
108 
109     /**
110      * The value of the <code>pos</code> field at the time the last
111      * <code>mark</code> method was called.
112      * <p>
113      * This value is always
114      * in the range <code>-1</code> through <code>pos</code>.
115      * If there is no marked position in  the input
116      * stream, this field is <code>-1</code>. If
117      * there is a marked position in the input
118      * stream,  then <code>buf[markpos]</code>
119      * is the first byte to be supplied as input
120      * after a <code>reset</code> operation. If
121      * <code>markpos</code> is not <code>-1</code>,
122      * then all bytes from positions <code>buf[markpos]</code>
123      * through  <code>buf[pos-1]</code> must remain
124      * in the buffer array (though they may be
125      * moved to  another place in the buffer array,
126      * with suitable adjustments to the values
127      * of <code>count</code>,  <code>pos</code>,
128      * and <code>markpos</code>); they may not
129      * be discarded unless and until the difference
130      * between <code>pos</code> and <code>markpos</code>
131      * exceeds <code>marklimit</code>.
132      *
133      * @see     java.io.BufferedInputStream#mark(int)
134      * @see     java.io.BufferedInputStream#pos
135      */
136     protected int markpos = -1;
137 
138     /**
139      * The maximum read ahead allowed after a call to the
140      * <code>mark</code> method before subsequent calls to the
141      * <code>reset</code> method fail.
142      * Whenever the difference between <code>pos</code>
143      * and <code>markpos</code> exceeds <code>marklimit</code>,
144      * then the  mark may be dropped by setting
145      * <code>markpos</code> to <code>-1</code>.
146      *
147      * @see     java.io.BufferedInputStream#mark(int)
148      * @see     java.io.BufferedInputStream#reset()
149      */
150     protected int marklimit;
151 
152     /**
153      * Check to make sure that underlying input stream has not been
154      * nulled out due to close; if not return it;
155      */
156     private InputStream getInIfOpen() throws IOException {
157         InputStream input = in;
158         if (input == null)
159             throw new IOException("Stream closed");
160         return input;
161     }
162 
163     /**
164      * Check to make sure that buffer has not been nulled out due to
165      * close; if not return it;
166      */
167     private byte[] getBufIfOpen() throws IOException {
168         byte[] buffer = buf;
169         if (buffer == null)
170             throw new IOException("Stream closed");
171         return buffer;
172     }
173 
174     /**
175      * Creates a <code>BufferedInputStream</code>
176      * and saves its  argument, the input stream
177      * <code>in</code>, for later use. An internal
178      * buffer array is created and  stored in <code>buf</code>.
179      *
180      * @param   in   the underlying input stream.
181      */
182     public BufferedInputStream(InputStream in) {
183         this(in, DEFAULT_BUFFER_SIZE);
184     }
185 
186     /**
187      * Creates a <code>BufferedInputStream</code>
188      * with the specified buffer size,
189      * and saves its  argument, the input stream
190      * <code>in</code>, for later use.  An internal
191      * buffer array of length  <code>size</code>
192      * is created and stored in <code>buf</code>.
193      *
194      * @param   in     the underlying input stream.
195      * @param   size   the buffer size.
196      * @exception IllegalArgumentException if {@code size <= 0}.
197      */
198     public BufferedInputStream(InputStream in, int size) {
199         super(in);
200         if (size <= 0) {
201             throw new IllegalArgumentException("Buffer size <= 0");
202         }
203         buf = new byte[size];
204     }
205 
206     /**
207      * Fills the buffer with more data, taking into account
208      * shuffling and other tricks for dealing with marks.
209      * Assumes that it is being called by a synchronized method.
210      * This method also assumes that all data has already been read in,
211      * hence pos > count.
212      */
213     private void fill() throws IOException {
214         byte[] buffer = getBufIfOpen();
215         if (markpos < 0)
216             pos = 0;            /* no mark: throw away the buffer */
217         else if (pos >= buffer.length)  /* no room left in buffer */
218             if (markpos > 0) {  /* can throw away early part of the buffer */
219                 int sz = pos - markpos;
220                 System.arraycopy(buffer, markpos, buffer, 0, sz);
221                 pos = sz;
222                 markpos = 0;
223             } else if (buffer.length >= marklimit) {
224                 markpos = -1;   /* buffer got too big, invalidate mark */
225                 pos = 0;        /* drop buffer contents */
226             } else if (buffer.length >= MAX_BUFFER_SIZE) {
227                 throw new OutOfMemoryError("Required array size too large");
228             } else {            /* grow buffer */
229                 int nsz = (pos <= MAX_BUFFER_SIZE - pos) ?
230                         pos * 2 : MAX_BUFFER_SIZE;
231                 if (nsz > marklimit)
232                     nsz = marklimit;
233                 byte nbuf[] = new byte[nsz];
234                 System.arraycopy(buffer, 0, nbuf, 0, pos);
235                 if (!bufUpdater.compareAndSet(this, buffer, nbuf)) {
236                     // Can't replace buf if there was an async close.
237                     // Note: This would need to be changed if fill()
238                     // is ever made accessible to multiple threads.
239                     // But for now, the only way CAS can fail is via close.
240                     // assert buf == null;
241                     throw new IOException("Stream closed");
242                 }
243                 buffer = nbuf;
244             }
245         count = pos;
246         int n = getInIfOpen().read(buffer, pos, buffer.length - pos);
247         if (n > 0)
248             count = n + pos;
249     }
250 
251     /**
252      * See
253      * the general contract of the <code>read</code>
254      * method of <code>InputStream</code>.
255      *
256      * @return     the next byte of data, or <code>-1</code> if the end of the
257      *             stream is reached.
258      * @exception  IOException  if this input stream has been closed by
259      *                          invoking its {@link #close()} method,
260      *                          or an I/O error occurs.
261      * @see        java.io.FilterInputStream#in
262      */
263     public synchronized int read() throws IOException {
264         if (pos >= count) {
265             fill();
266             if (pos >= count)
267                 return -1;
268         }
269         return getBufIfOpen()[pos++] & 0xff;
270     }
271 
272     /**
273      * Read characters into a portion of an array, reading from the underlying
274      * stream at most once if necessary.
275      */
276     private int read1(byte[] b, int off, int len) throws IOException {
277         int avail = count - pos;
278         if (avail <= 0) {
279             /* If the requested length is at least as large as the buffer, and
280                if there is no mark/reset activity, do not bother to copy the
281                bytes into the local buffer.  In this way buffered streams will
282                cascade harmlessly. */
283             if (len >= getBufIfOpen().length && markpos < 0) {
284                 return getInIfOpen().read(b, off, len);
285             }
286             fill();
287             avail = count - pos;
288             if (avail <= 0) return -1;
289         }
290         int cnt = (avail < len) ? avail : len;
291         System.arraycopy(getBufIfOpen(), pos, b, off, cnt);
292         pos += cnt;
293         return cnt;
294     }
295 
296     /**
297      * Reads bytes from this byte-input stream into the specified byte array,
298      * starting at the given offset.
299      *
300      * <p> This method implements the general contract of the corresponding
301      * <code>{@link InputStream#read(byte[], int, int) read}</code> method of
302      * the <code>{@link InputStream}</code> class.  As an additional
303      * convenience, it attempts to read as many bytes as possible by repeatedly
304      * invoking the <code>read</code> method of the underlying stream.  This
305      * iterated <code>read</code> continues until one of the following
306      * conditions becomes true: <ul>
307      *
308      *   <li> The specified number of bytes have been read,
309      *
310      *   <li> The <code>read</code> method of the underlying stream returns
311      *   <code>-1</code>, indicating end-of-file, or
312      *
313      *   <li> The <code>available</code> method of the underlying stream
314      *   returns zero, indicating that further input requests would block.
315      *
316      * </ul> If the first <code>read</code> on the underlying stream returns
317      * <code>-1</code> to indicate end-of-file then this method returns
318      * <code>-1</code>.  Otherwise this method returns the number of bytes
319      * actually read.
320      *
321      * <p> Subclasses of this class are encouraged, but not required, to
322      * attempt to read as many bytes as possible in the same fashion.
323      *
324      * @param      b     destination buffer.
325      * @param      off   offset at which to start storing bytes.
326      * @param      len   maximum number of bytes to read.
327      * @return     the number of bytes read, or <code>-1</code> if the end of
328      *             the stream has been reached.
329      * @exception  IOException  if this input stream has been closed by
330      *                          invoking its {@link #close()} method,
331      *                          or an I/O error occurs.
332      */
333     public synchronized int read(byte b[], int off, int len)
334         throws IOException
335     {
336         getBufIfOpen(); // Check for closed stream
337         if ((off | len | (off + len) | (b.length - (off + len))) < 0) {
338             throw new IndexOutOfBoundsException();
339         } else if (len == 0) {
340             return 0;
341         }
342 
343         int n = 0;
344         for (;;) {
345             int nread = read1(b, off + n, len - n);
346             if (nread <= 0)
347                 return (n == 0) ? nread : n;
348             n += nread;
349             if (n >= len)
350                 return n;
351             // if not closed but no bytes available, return
352             InputStream input = in;
353             if (input != null && input.available() <= 0)
354                 return n;
355         }
356     }
357 
358     /**
359      * See the general contract of the <code>skip</code>
360      * method of <code>InputStream</code>.
361      *
362      * @exception  IOException  if the stream does not support seek,
363      *                          or if this input stream has been closed by
364      *                          invoking its {@link #close()} method, or an
365      *                          I/O error occurs.
366      */
367     public synchronized long skip(long n) throws IOException {
368         getBufIfOpen(); // Check for closed stream
369         if (n <= 0) {
370             return 0;
371         }
372         long avail = count - pos;
373 
374         if (avail <= 0) {
375             // If no mark position set then don't keep in buffer
376             if (markpos <0)
377                 return getInIfOpen().skip(n);
378 
379             // Fill in buffer to save bytes for reset
380             fill();
381             avail = count - pos;
382             if (avail <= 0)
383                 return 0;
384         }
385 
386         long skipped = (avail < n) ? avail : n;
387         pos += skipped;
388         return skipped;
389     }
390 
391     /**
392      * Returns an estimate of the number of bytes that can be read (or
393      * skipped over) from this input stream without blocking by the next
394      * invocation of a method for this input stream. The next invocation might be
395      * the same thread or another thread.  A single read or skip of this
396      * many bytes will not block, but may read or skip fewer bytes.
397      * <p>
398      * This method returns the sum of the number of bytes remaining to be read in
399      * the buffer (<code>count&nbsp;- pos</code>) and the result of calling the
400      * {@link java.io.FilterInputStream#in in}.available().
401      *
402      * @return     an estimate of the number of bytes that can be read (or skipped
403      *             over) from this input stream without blocking.
404      * @exception  IOException  if this input stream has been closed by
405      *                          invoking its {@link #close()} method,
406      *                          or an I/O error occurs.
407      */
408     public synchronized int available() throws IOException {
409         int n = count - pos;
410         int avail = getInIfOpen().available();
411         return n > (Integer.MAX_VALUE - avail)
412                     ? Integer.MAX_VALUE
413                     : n + avail;
414     }
415 
416     /**
417      * See the general contract of the <code>mark</code>
418      * method of <code>InputStream</code>.
419      *
420      * @param   readlimit   the maximum limit of bytes that can be read before
421      *                      the mark position becomes invalid.
422      * @see     java.io.BufferedInputStream#reset()
423      */
424     public synchronized void mark(int readlimit) {
425         marklimit = readlimit;
426         markpos = pos;
427     }
428 
429     /**
430      * See the general contract of the <code>reset</code>
431      * method of <code>InputStream</code>.
432      * <p>
433      * If <code>markpos</code> is <code>-1</code>
434      * (no mark has been set or the mark has been
435      * invalidated), an <code>IOException</code>
436      * is thrown. Otherwise, <code>pos</code> is
437      * set equal to <code>markpos</code>.
438      *
439      * @exception  IOException  if this stream has not been marked or,
440      *                  if the mark has been invalidated, or the stream
441      *                  has been closed by invoking its {@link #close()}
442      *                  method, or an I/O error occurs.
443      * @see        java.io.BufferedInputStream#mark(int)
444      */
445     public synchronized void reset() throws IOException {
446         getBufIfOpen(); // Cause exception if closed
447         if (markpos < 0)
448             throw new IOException("Resetting to invalid mark");
449         pos = markpos;
450     }
451 
452     /**
453      * Tests if this input stream supports the <code>mark</code>
454      * and <code>reset</code> methods. The <code>markSupported</code>
455      * method of <code>BufferedInputStream</code> returns
456      * <code>true</code>.
457      *
458      * @return  a <code>boolean</code> indicating if this stream type supports
459      *          the <code>mark</code> and <code>reset</code> methods.
460      * @see     java.io.InputStream#mark(int)
461      * @see     java.io.InputStream#reset()
462      */
463     public boolean markSupported() {
464         return true;
465     }
466 
467     /**
468      * Closes this input stream and releases any system resources
469      * associated with the stream.
470      * Once the stream has been closed, further read(), available(), reset(),
471      * or skip() invocations will throw an IOException.
472      * Closing a previously closed stream has no effect.
473      *
474      * @exception  IOException  if an I/O error occurs.
475      */
476     public void close() throws IOException {
477         byte[] buffer;
478         while ( (buffer = buf) != null) {
479             if (bufUpdater.compareAndSet(this, buffer, null)) {
480                 InputStream input = in;
481                 in = null;
482                 if (input != null)
483                     input.close();
484                 return;
485             }
486             // Else retry in case a new buf was CASed in fill()
487         }
488     }
489 }