1 /*
2 * Licensed to the Apache Software Foundation (ASF) under one or more
3 * contributor license agreements. See the NOTICE file distributed with
4 * this work for additional information regarding copyright ownership.
5 * The ASF licenses this file to You under the Apache License, Version 2.0
6 * (the "License"); you may not use this file except in compliance with
7 * the License. You may obtain a copy of the License at
8 *
9 * http://www.apache.org/licenses/LICENSE-2.0
10 *
11 * Unless required by applicable law or agreed to in writing, software
12 * distributed under the License is distributed on an "AS IS" BASIS,
13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 * See the License for the specific language governing permissions and
15 * limitations under the License.
16 */
17
18 package org.apache.commons.io.input;
19
20 import java.io.BufferedInputStream;
21 import java.io.IOException;
22 import java.io.InputStream;
23
24 import org.apache.commons.io.IOUtils;
25 import org.apache.commons.io.build.AbstractStreamBuilder;
26
27 /**
28 * An unsynchronized version of {@link BufferedInputStream}, not thread-safe.
29 * <p>
30 * Wraps an existing {@link InputStream} and <em>buffers</em> the input. Expensive interaction with the underlying input stream is minimized, since most
31 * (smaller) requests can be satisfied by accessing the buffer alone. The drawback is that some extra space is required to hold the buffer and that copying
32 * takes place when filling that buffer, but this is usually outweighed by the performance benefits.
33 * </p>
34 * <p>
35 * To build an instance, use {@link Builder}.
36 * </p>
37 * <p>
38 * A typical application pattern for the class looks like this:
39 * </p>
40 *
41 * <pre>
42 * UnsynchronizedBufferedInputStream s = new UnsynchronizedBufferedInputStream.Builder().
43 * .setInputStream(new FileInputStream("file.java"))
44 * .setBufferSize(8192)
45 * .get();
46 * </pre>
47 * <p>
48 * Provenance: Apache Harmony and modified.
49 * </p>
50 *
51 * @see Builder
52 * @see BufferedInputStream
53 * @since 2.12.0
54 */
55 //@NotThreadSafe
56 public final class UnsynchronizedBufferedInputStream extends UnsynchronizedFilterInputStream {
57
58 // @formatter:off
59 /**
60 * Builds a new {@link UnsynchronizedBufferedInputStream}.
61 *
62 * <p>
63 * Using File IO:
64 * </p>
65 * <pre>{@code
66 * UnsynchronizedBufferedInputStream s = UnsynchronizedBufferedInputStream.builder()
67 * .setFile(file)
68 * .setBufferSize(8192)
69 * .get();}
70 * </pre>
71 * <p>
72 * Using NIO Path:
73 * </p>
74 * <pre>{@code
75 * UnsynchronizedBufferedInputStream s = UnsynchronizedBufferedInputStream.builder()
76 * .setPath(path)
77 * .setBufferSize(8192)
78 * .get();}
79 * </pre>
80 *
81 * @see #get()
82 */
83 // @formatter:on
84 public static class Builder extends AbstractStreamBuilder<UnsynchronizedBufferedInputStream, Builder> {
85
86 /**
87 * Builds a new {@link UnsynchronizedBufferedInputStream}.
88 * <p>
89 * You must set input that supports {@link #getInputStream()} on this builder, otherwise, this method throws an exception.
90 * </p>
91 * <p>
92 * This builder use the following aspects:
93 * </p>
94 * <ul>
95 * <li>{@link #getInputStream()}</li>
96 * <li>{@link #getBufferSize()}</li>
97 * </ul>
98 *
99 * @return a new instance.
100 * @throws IllegalStateException if the {@code origin} is {@code null}.
101 * @throws UnsupportedOperationException if the origin cannot be converted to an {@link InputStream}.
102 * @throws IOException if an I/O error occurs.
103 * @see #getInputStream()
104 * @see #getBufferSize()
105 */
106 @SuppressWarnings("resource") // Caller closes.
107 @Override
108 public UnsynchronizedBufferedInputStream get() throws IOException {
109 return new UnsynchronizedBufferedInputStream(getInputStream(), getBufferSize());
110 }
111
112 }
113
114 /**
115 * The buffer containing the current bytes read from the target InputStream.
116 */
117 protected volatile byte[] buffer;
118
119 /**
120 * The total number of bytes inside the byte array {@code buffer}.
121 */
122 protected int count;
123
124 /**
125 * The current limit, which when passed, invalidates the current mark.
126 */
127 protected int markLimit;
128
129 /**
130 * The currently marked position. -1 indicates no mark has been set or the mark has been invalidated.
131 */
132 protected int markPos = IOUtils.EOF;
133
134 /**
135 * The current position within the byte array {@code buffer}.
136 */
137 protected int pos;
138
139 /**
140 * Constructs a new {@code BufferedInputStream} on the {@link InputStream} {@code in}. The buffer size is specified by the parameter {@code size} and all
141 * reads are now filtered through this stream.
142 *
143 * @param in the input stream the buffer reads from.
144 * @param size the size of buffer to allocate.
145 * @throws IllegalArgumentException if {@code size < 0}.
146 */
147 private UnsynchronizedBufferedInputStream(final InputStream in, final int size) {
148 super(in);
149 if (size <= 0) {
150 throw new IllegalArgumentException("Size must be > 0");
151 }
152 buffer = new byte[size];
153 }
154
155 /**
156 * Returns the number of bytes that are available before this stream will block. This method returns the number of bytes available in the buffer plus those
157 * available in the source stream.
158 *
159 * @return the number of bytes available before blocking.
160 * @throws IOException if this stream is closed.
161 */
162 @Override
163 public int available() throws IOException {
164 final InputStream localIn = inputStream; // 'in' could be invalidated by close()
165 if (buffer == null || localIn == null) {
166 throw new IOException("Stream is closed");
167 }
168 return count - pos + localIn.available();
169 }
170
171 /**
172 * Closes this stream. The source stream is closed and any resources associated with it are released.
173 *
174 * @throws IOException if an error occurs while closing this stream.
175 */
176 @Override
177 public void close() throws IOException {
178 buffer = null;
179 final InputStream localIn = inputStream;
180 inputStream = null;
181 if (localIn != null) {
182 localIn.close();
183 }
184 }
185
186 private int fillBuffer(final InputStream localIn, byte[] localBuf) throws IOException {
187 if (markPos == IOUtils.EOF || pos - markPos >= markLimit) {
188 /* Mark position not set or exceeded readLimit */
189 final int result = localIn.read(localBuf);
190 if (result > 0) {
191 markPos = IOUtils.EOF;
192 pos = 0;
193 count = result;
194 }
195 return result;
196 }
197 if (markPos == 0 && markLimit > localBuf.length) {
198 /* Increase buffer size to accommodate the readLimit */
199 int newLength = localBuf.length * 2;
200 if (newLength > markLimit) {
201 newLength = markLimit;
202 }
203 final byte[] newbuf = new byte[newLength];
204 System.arraycopy(localBuf, 0, newbuf, 0, localBuf.length);
205 // Reassign buffer, which will invalidate any local references
206 // FIXME: what if buffer was null?
207 localBuf = buffer = newbuf;
208 } else if (markPos > 0) {
209 System.arraycopy(localBuf, markPos, localBuf, 0, localBuf.length - markPos);
210 }
211 // Set the new position and mark position
212 pos -= markPos;
213 count = markPos = 0;
214 final int bytesread = localIn.read(localBuf, pos, localBuf.length - pos);
215 count = bytesread <= 0 ? pos : pos + bytesread;
216 return bytesread;
217 }
218
219 byte[] getBuffer() {
220 return buffer;
221 }
222
223 /**
224 * Sets a mark position in this stream. The parameter {@code readLimit} indicates how many bytes can be read before a mark is invalidated. Calling
225 * {@code reset()} will reposition the stream back to the marked position if {@code readLimit} has not been surpassed. The underlying buffer may be
226 * increased in size to allow {@code readLimit} number of bytes to be supported.
227 *
228 * @param readLimit the number of bytes that can be read before the mark is invalidated.
229 * @see #reset()
230 */
231 @Override
232 public void mark(final int readLimit) {
233 markLimit = readLimit;
234 markPos = pos;
235 }
236
237 /**
238 * Indicates whether {@code BufferedInputStream} supports the {@code mark()} and {@code reset()} methods.
239 *
240 * @return {@code true} for BufferedInputStreams.
241 * @see #mark(int)
242 * @see #reset()
243 */
244 @Override
245 public boolean markSupported() {
246 return true;
247 }
248
249 /**
250 * Reads a single byte from this stream and returns it as an integer in the range from 0 to 255. Returns -1 if the end of the source string has been
251 * reached. If the internal buffer does not contain any available bytes then it is filled from the source stream and the first byte is returned.
252 *
253 * @return the byte read or -1 if the end of the source stream has been reached.
254 * @throws IOException if this stream is closed or another IOException occurs.
255 */
256 @Override
257 public int read() throws IOException {
258 // Use local refs since buf and in may be invalidated by an
259 // unsynchronized close()
260 byte[] localBuf = buffer;
261 final InputStream localIn = inputStream;
262 if (localBuf == null || localIn == null) {
263 throw new IOException("Stream is closed");
264 }
265
266 /* Are there buffered bytes available? */
267 if (pos >= count && fillBuffer(localIn, localBuf) == IOUtils.EOF) {
268 return IOUtils.EOF; /* no, fill buffer */
269 }
270 // localBuf may have been invalidated by fillbuf
271 if (localBuf != buffer) {
272 localBuf = buffer;
273 if (localBuf == null) {
274 throw new IOException("Stream is closed");
275 }
276 }
277
278 /* Did filling the buffer fail with -1 (EOF)? */
279 if (count - pos > 0) {
280 return localBuf[pos++] & 0xFF;
281 }
282 return IOUtils.EOF;
283 }
284
285 /**
286 * Reads at most {@code length} bytes from this stream and stores them in byte array {@code buffer} starting at offset {@code offset}. Returns the number of
287 * bytes actually read or -1 if no bytes were read and the end of the stream was encountered. If all the buffered bytes have been used, a mark has not been
288 * set and the requested number of bytes is larger than the receiver's buffer size, this implementation bypasses the buffer and simply places the results
289 * directly into {@code buffer}.
290 *
291 * @param dest the byte array in which to store the bytes read.
292 * @param offset the initial position in {@code buffer} to store the bytes read from this stream.
293 * @param length the maximum number of bytes to store in {@code buffer}.
294 * @return the number of bytes actually read or -1 if end of stream.
295 * @throws IndexOutOfBoundsException if {@code offset < 0} or {@code length < 0}, or if {@code offset + length} is greater than the size of {@code buffer}.
296 * @throws IOException if the stream is already closed or another IOException occurs.
297 */
298 @Override
299 public int read(final byte[] dest, int offset, final int length) throws IOException {
300 // Use local ref since buf may be invalidated by an unsynchronized
301 // close()
302 byte[] localBuf = buffer;
303 if (localBuf == null) {
304 throw new IOException("Stream is closed");
305 }
306 // avoid int overflow
307 if (offset > dest.length - length || offset < 0 || length < 0) {
308 throw new IndexOutOfBoundsException();
309 }
310 if (length == 0) {
311 return 0;
312 }
313 final InputStream localIn = inputStream;
314 if (localIn == null) {
315 throw new IOException("Stream is closed");
316 }
317
318 int required;
319 if (pos < count) {
320 /* There are bytes available in the buffer. */
321 final int copylength = count - pos >= length ? length : count - pos;
322 System.arraycopy(localBuf, pos, dest, offset, copylength);
323 pos += copylength;
324 if (copylength == length || localIn.available() == 0) {
325 return copylength;
326 }
327 offset += copylength;
328 required = length - copylength;
329 } else {
330 required = length;
331 }
332
333 while (true) {
334 final int read;
335 /*
336 * If we're not marked and the required size is greater than the buffer, simply read the bytes directly bypassing the buffer.
337 */
338 if (markPos == IOUtils.EOF && required >= localBuf.length) {
339 read = localIn.read(dest, offset, required);
340 if (read == IOUtils.EOF) {
341 return required == length ? IOUtils.EOF : length - required;
342 }
343 } else {
344 if (fillBuffer(localIn, localBuf) == IOUtils.EOF) {
345 return required == length ? IOUtils.EOF : length - required;
346 }
347 // localBuf may have been invalidated by fillBuffer()
348 if (localBuf != buffer) {
349 localBuf = buffer;
350 if (localBuf == null) {
351 throw new IOException("Stream is closed");
352 }
353 }
354
355 read = count - pos >= required ? required : count - pos;
356 System.arraycopy(localBuf, pos, dest, offset, read);
357 pos += read;
358 }
359 required -= read;
360 if (required == 0) {
361 return length;
362 }
363 if (localIn.available() == 0) {
364 return length - required;
365 }
366 offset += read;
367 }
368 }
369
370 /**
371 * Resets this stream to the last marked location.
372 *
373 * @throws IOException if this stream is closed, no mark has been set or the mark is no longer valid because more than {@code readLimit} bytes have been
374 * read since setting the mark.
375 * @see #mark(int)
376 */
377 @Override
378 public void reset() throws IOException {
379 if (buffer == null) {
380 throw new IOException("Stream is closed");
381 }
382 if (IOUtils.EOF == markPos) {
383 throw new IOException("Mark has been invalidated");
384 }
385 pos = markPos;
386 }
387
388 /**
389 * Skips {@code amount} number of bytes in this stream. Subsequent {@code read()}'s will not return these bytes unless {@code reset()} is used.
390 *
391 * @param amount the number of bytes to skip. {@code skip} does nothing and returns 0 if {@code amount} is less than zero.
392 * @return the number of bytes actually skipped.
393 * @throws IOException if this stream is closed or another IOException occurs.
394 */
395 @Override
396 public long skip(final long amount) throws IOException {
397 // Use local refs since buf and in may be invalidated by an
398 // unsynchronized close()
399 final byte[] localBuf = buffer;
400 final InputStream localIn = inputStream;
401 if (localBuf == null) {
402 throw new IOException("Stream is closed");
403 }
404 if (amount < 1) {
405 return 0;
406 }
407 if (localIn == null) {
408 throw new IOException("Stream is closed");
409 }
410
411 if (count - pos >= amount) {
412 // (int count - int pos) here is always an int so amount is also in the int range if the above test is true.
413 // We can safely cast to int and avoid static analysis warnings.
414 pos += (int) amount;
415 return amount;
416 }
417 int read = count - pos;
418 pos = count;
419
420 if (markPos != IOUtils.EOF && amount <= markLimit) {
421 if (fillBuffer(localIn, localBuf) == IOUtils.EOF) {
422 return read;
423 }
424 if (count - pos >= amount - read) {
425 // (int count - int pos) here is always an int so (amount - read) is also in the int range if the above test is true.
426 // We can safely cast to int and avoid static analysis warnings.
427 pos += (int) amount - read;
428 return amount;
429 }
430 // Couldn't get all the bytes, skip what we read
431 read += count - pos;
432 pos = count;
433 return read;
434 }
435 return read + localIn.skip(amount - read);
436 }
437 }