001/* 002 * Licensed to the Apache Software Foundation (ASF) under one or more 003 * contributor license agreements. See the NOTICE file distributed with 004 * this work for additional information regarding copyright ownership. 005 * The ASF licenses this file to You under the Apache License, Version 2.0 006 * (the "License"); you may not use this file except in compliance with 007 * the License. You may obtain a copy of the License at 008 * 009 * http://www.apache.org/licenses/LICENSE-2.0 010 * 011 * Unless required by applicable law or agreed to in writing, software 012 * distributed under the License is distributed on an "AS IS" BASIS, 013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 014 * See the License for the specific language governing permissions and 015 * limitations under the License. 016 */ 017 018package org.apache.commons.io.input; 019 020import java.io.BufferedInputStream; 021import java.io.IOException; 022import java.io.InputStream; 023 024import org.apache.commons.io.IOUtils; 025import org.apache.commons.io.build.AbstractStreamBuilder; 026 027/** 028 * An unsynchronized version of {@link BufferedInputStream}, not thread-safe. 029 * <p> 030 * Wraps an existing {@link InputStream} and <em>buffers</em> the input. Expensive interaction with the underlying input stream is minimized, since most 031 * (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 032 * takes place when filling that buffer, but this is usually outweighed by the performance benefits. 033 * </p> 034 * <p> 035 * To build an instance, use {@link Builder}. 036 * </p> 037 * <p> 038 * A typical application pattern for the class looks like this: 039 * </p> 040 * 041 * <pre> 042 * UnsynchronizedBufferedInputStream s = new UnsynchronizedBufferedInputStream.Builder(). 043 * .setInputStream(new FileInputStream("file.java")) 044 * .setBufferSize(8192) 045 * .get(); 046 * </pre> 047 * <p> 048 * Provenance: Apache Harmony and modified. 049 * </p> 050 * 051 * @see Builder 052 * @see BufferedInputStream 053 * @since 2.12.0 054 */ 055//@NotThreadSafe 056public final class UnsynchronizedBufferedInputStream extends UnsynchronizedFilterInputStream { 057 058 // @formatter:off 059 /** 060 * Builds a new {@link UnsynchronizedBufferedInputStream}. 061 * 062 * <p> 063 * Using File IO: 064 * </p> 065 * <pre>{@code 066 * UnsynchronizedBufferedInputStream s = UnsynchronizedBufferedInputStream.builder() 067 * .setFile(file) 068 * .setBufferSize(8192) 069 * .get();} 070 * </pre> 071 * <p> 072 * Using NIO Path: 073 * </p> 074 * <pre>{@code 075 * UnsynchronizedBufferedInputStream s = UnsynchronizedBufferedInputStream.builder() 076 * .setPath(path) 077 * .setBufferSize(8192) 078 * .get();} 079 * </pre> 080 * 081 * @see #get() 082 */ 083 // @formatter:on 084 public static class Builder extends AbstractStreamBuilder<UnsynchronizedBufferedInputStream, Builder> { 085 086 /** 087 * Constructs a builder of {@link UnsynchronizedBufferedInputStream}. 088 */ 089 public Builder() { 090 // empty 091 } 092 093 /** 094 * Builds a new {@link UnsynchronizedBufferedInputStream}. 095 * <p> 096 * You must set an aspect that supports {@link #getInputStream()} on this builder, otherwise, this method throws an exception. 097 * </p> 098 * <p> 099 * This builder uses the following aspects: 100 * </p> 101 * <ul> 102 * <li>{@link #getInputStream()}</li> 103 * <li>{@link #getBufferSize()}</li> 104 * </ul> 105 * 106 * @return a new instance. 107 * @throws IllegalStateException if the {@code origin} is {@code null}. 108 * @throws UnsupportedOperationException if the origin cannot be converted to an {@link InputStream}. 109 * @throws IOException if an I/O error occurs converting to an {@link InputStream} using {@link #getInputStream()}. 110 * @see #getInputStream() 111 * @see #getBufferSize() 112 * @see #getUnchecked() 113 */ 114 @Override 115 public UnsynchronizedBufferedInputStream get() throws IOException { 116 return new UnsynchronizedBufferedInputStream(getInputStream(), getBufferSize()); 117 } 118 119 } 120 121 /** 122 * The buffer containing the current bytes read from the target InputStream. 123 */ 124 protected volatile byte[] buffer; 125 126 /** 127 * The total number of bytes inside the byte array {@code buffer}. 128 */ 129 protected int count; 130 131 /** 132 * The current limit, which when passed, invalidates the current mark. 133 */ 134 protected int markLimit; 135 136 /** 137 * The currently marked position. -1 indicates no mark has been set or the mark has been invalidated. 138 */ 139 protected int markPos = IOUtils.EOF; 140 141 /** 142 * The current position within the byte array {@code buffer}. 143 */ 144 protected int pos; 145 146 /** 147 * Constructs a new {@code BufferedInputStream} on the {@link InputStream} {@code in}. The buffer size is specified by the parameter {@code size} and all 148 * reads are now filtered through this stream. 149 * 150 * @param in the input stream the buffer reads from. 151 * @param size the size of buffer to allocate. 152 * @throws IllegalArgumentException if {@code size < 0}. 153 */ 154 private UnsynchronizedBufferedInputStream(final InputStream in, final int size) { 155 super(in); 156 if (size <= 0) { 157 throw new IllegalArgumentException("Size must be > 0"); 158 } 159 buffer = new byte[size]; 160 } 161 162 /** 163 * 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 164 * available in the source stream. 165 * 166 * @return the number of bytes available before blocking. 167 * @throws IOException if this stream is closed. 168 */ 169 @Override 170 public int available() throws IOException { 171 final InputStream localIn = inputStream; // 'in' could be invalidated by close() 172 if (buffer == null || localIn == null) { 173 throw new IOException("Stream is closed"); 174 } 175 return count - pos + localIn.available(); 176 } 177 178 /** 179 * Closes this stream. The source stream is closed and any resources associated with it are released. 180 * 181 * @throws IOException if an error occurs while closing this stream. 182 */ 183 @Override 184 public void close() throws IOException { 185 buffer = null; 186 final InputStream localIn = inputStream; 187 inputStream = null; 188 if (localIn != null) { 189 localIn.close(); 190 } 191 } 192 193 private int fillBuffer(final InputStream localIn, byte[] localBuf) throws IOException { 194 if (markPos == IOUtils.EOF || pos - markPos >= markLimit) { 195 /* Mark position not set or exceeded readLimit */ 196 final int result = localIn.read(localBuf); 197 if (result > 0) { 198 markPos = IOUtils.EOF; 199 pos = 0; 200 count = result; 201 } 202 return result; 203 } 204 if (markPos == 0 && markLimit > localBuf.length) { 205 /* Increase buffer size to accommodate the readLimit */ 206 int newLength = localBuf.length * 2; 207 if (newLength > markLimit) { 208 newLength = markLimit; 209 } 210 final byte[] newbuf = new byte[newLength]; 211 System.arraycopy(localBuf, 0, newbuf, 0, localBuf.length); 212 // Reassign buffer, which will invalidate any local references 213 // FIXME: what if buffer was null? 214 localBuf = buffer = newbuf; 215 } else if (markPos > 0) { 216 System.arraycopy(localBuf, markPos, localBuf, 0, localBuf.length - markPos); 217 } 218 // Set the new position and mark position 219 pos -= markPos; 220 count = markPos = 0; 221 final int bytesread = localIn.read(localBuf, pos, localBuf.length - pos); 222 count = bytesread <= 0 ? pos : pos + bytesread; 223 return bytesread; 224 } 225 226 byte[] getBuffer() { 227 return buffer; 228 } 229 230 /** 231 * Sets a mark position in this stream. The parameter {@code readLimit} indicates how many bytes can be read before a mark is invalidated. Calling 232 * {@code reset()} will reposition the stream back to the marked position if {@code readLimit} has not been surpassed. The underlying buffer may be 233 * increased in size to allow {@code readLimit} number of bytes to be supported. 234 * 235 * @param readLimit the number of bytes that can be read before the mark is invalidated. 236 * @see #reset() 237 */ 238 @Override 239 public void mark(final int readLimit) { 240 markLimit = readLimit; 241 markPos = pos; 242 } 243 244 /** 245 * Indicates whether {@code BufferedInputStream} supports the {@code mark()} and {@code reset()} methods. 246 * 247 * @return {@code true} for BufferedInputStreams. 248 * @see #mark(int) 249 * @see #reset() 250 */ 251 @Override 252 public boolean markSupported() { 253 return true; 254 } 255 256 /** 257 * 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 258 * 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. 259 * 260 * @return the byte read or -1 if the end of the source stream has been reached. 261 * @throws IOException if this stream is closed or another IOException occurs. 262 */ 263 @Override 264 public int read() throws IOException { 265 // Use local refs since buf and in may be invalidated by an 266 // unsynchronized close() 267 byte[] localBuf = buffer; 268 final InputStream localIn = inputStream; 269 if (localBuf == null || localIn == null) { 270 throw new IOException("Stream is closed"); 271 } 272 273 /* Are there buffered bytes available? */ 274 if (pos >= count && fillBuffer(localIn, localBuf) == IOUtils.EOF) { 275 return IOUtils.EOF; /* no, fill buffer */ 276 } 277 // localBuf may have been invalidated by fillbuf 278 if (localBuf != buffer) { 279 localBuf = buffer; 280 if (localBuf == null) { 281 throw new IOException("Stream is closed"); 282 } 283 } 284 285 /* Did filling the buffer fail with -1 (EOF)? */ 286 if (count - pos > 0) { 287 return localBuf[pos++] & 0xFF; 288 } 289 return IOUtils.EOF; 290 } 291 292 /** 293 * 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 294 * 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 295 * 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 296 * directly into {@code buffer}. 297 * 298 * @param dest the byte array in which to store the bytes read. 299 * @param offset the initial position in {@code buffer} to store the bytes read from this stream. 300 * @param length the maximum number of bytes to store in {@code buffer}. 301 * @return the number of bytes actually read or -1 if end of stream. 302 * @throws IndexOutOfBoundsException if {@code offset < 0} or {@code length < 0}, or if {@code offset + length} is greater than the size of {@code buffer}. 303 * @throws IOException if the stream is already closed or another IOException occurs. 304 */ 305 @Override 306 public int read(final byte[] dest, int offset, final int length) throws IOException { 307 // Use local ref since buf may be invalidated by an unsynchronized 308 // close() 309 byte[] localBuf = buffer; 310 if (localBuf == null) { 311 throw new IOException("Stream is closed"); 312 } 313 // avoid int overflow 314 if (offset > dest.length - length || offset < 0 || length < 0) { 315 throw new IndexOutOfBoundsException(); 316 } 317 if (length == 0) { 318 return 0; 319 } 320 final InputStream localIn = inputStream; 321 if (localIn == null) { 322 throw new IOException("Stream is closed"); 323 } 324 325 int required; 326 if (pos < count) { 327 /* There are bytes available in the buffer. */ 328 final int copylength = count - pos >= length ? length : count - pos; 329 System.arraycopy(localBuf, pos, dest, offset, copylength); 330 pos += copylength; 331 if (copylength == length || localIn.available() == 0) { 332 return copylength; 333 } 334 offset += copylength; 335 required = length - copylength; 336 } else { 337 required = length; 338 } 339 340 while (true) { 341 final int read; 342 /* 343 * If we're not marked and the required size is greater than the buffer, simply read the bytes directly bypassing the buffer. 344 */ 345 if (markPos == IOUtils.EOF && required >= localBuf.length) { 346 read = localIn.read(dest, offset, required); 347 if (read == IOUtils.EOF) { 348 return required == length ? IOUtils.EOF : length - required; 349 } 350 } else { 351 if (fillBuffer(localIn, localBuf) == IOUtils.EOF) { 352 return required == length ? IOUtils.EOF : length - required; 353 } 354 // localBuf may have been invalidated by fillBuffer() 355 if (localBuf != buffer) { 356 localBuf = buffer; 357 if (localBuf == null) { 358 throw new IOException("Stream is closed"); 359 } 360 } 361 362 read = count - pos >= required ? required : count - pos; 363 System.arraycopy(localBuf, pos, dest, offset, read); 364 pos += read; 365 } 366 required -= read; 367 if (required == 0) { 368 return length; 369 } 370 if (localIn.available() == 0) { 371 return length - required; 372 } 373 offset += read; 374 } 375 } 376 377 /** 378 * Resets this stream to the last marked location. 379 * 380 * @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 381 * read since setting the mark. 382 * @see #mark(int) 383 */ 384 @Override 385 public void reset() throws IOException { 386 if (buffer == null) { 387 throw new IOException("Stream is closed"); 388 } 389 if (IOUtils.EOF == markPos) { 390 throw new IOException("Mark has been invalidated"); 391 } 392 pos = markPos; 393 } 394 395 /** 396 * Skips {@code amount} number of bytes in this stream. Subsequent {@code read()}'s will not return these bytes unless {@code reset()} is used. 397 * 398 * @param amount the number of bytes to skip. {@code skip} does nothing and returns 0 if {@code amount} is less than zero. 399 * @return the number of bytes actually skipped. 400 * @throws IOException if this stream is closed or another IOException occurs. 401 */ 402 @Override 403 public long skip(final long amount) throws IOException { 404 // Use local refs since buf and in may be invalidated by an 405 // unsynchronized close() 406 final byte[] localBuf = buffer; 407 final InputStream localIn = inputStream; 408 if (localBuf == null) { 409 throw new IOException("Stream is closed"); 410 } 411 if (amount < 1) { 412 return 0; 413 } 414 if (localIn == null) { 415 throw new IOException("Stream is closed"); 416 } 417 418 if (count - pos >= amount) { 419 // (int count - int pos) here is always an int so amount is also in the int range if the above test is true. 420 // We can safely cast to int and avoid static analysis warnings. 421 pos += (int) amount; 422 return amount; 423 } 424 int read = count - pos; 425 pos = count; 426 427 if (markPos != IOUtils.EOF && amount <= markLimit) { 428 if (fillBuffer(localIn, localBuf) == IOUtils.EOF) { 429 return read; 430 } 431 if (count - pos >= amount - read) { 432 // (int count - int pos) here is always an int so (amount - read) is also in the int range if the above test is true. 433 // We can safely cast to int and avoid static analysis warnings. 434 pos += (int) amount - read; 435 return amount; 436 } 437 // Couldn't get all the bytes, skip what we read 438 read += count - pos; 439 pos = count; 440 return read; 441 } 442 return read + localIn.skip(amount - read); 443 } 444}