streams/BufferedInputStream.java

   1 package be.nikiroo.utils.streams;
   2
   3 import java.io.IOException;
   4 import java.io.InputStream;
   5 import java.util.AbstractMap;
   6 import java.util.ArrayList;
   7 import java.util.List;
   8 import java.util.Map.Entry;
   9
  10 import be.nikiroo.utils.StringUtils;
  11
  12 /**
  13  * A simple {@link InputStream} that is buffered with a bytes array.
  14  * <p>
  15  * It is mostly intended to be used as a base class to create new
  16  * {@link InputStream}s with special operation modes, and to give some default
  17  * methods.
  18  *
  19  * @author niki
  20  */
  21 public class BufferedInputStream extends InputStream {
  22         /**
  23          * The size of the internal buffer (can be different if you pass your own
  24          * buffer, of course, and can also expand to search for longer "startsWith"
  25          * data).
  26          * <p>
  27          * Note that special "push-back" buffers can also be created during the life
  28          * of this stream.
  29          */
  30         static private final int BUFFER_SIZE = 4096;
  31
  32         /** The current position in the buffer. */
  33         protected int start;
  34         /** The index of the last usable position of the buffer. */
  35         protected int stop;
  36         /** The buffer itself. */
  37         protected byte[] buffer;
  38         /** An End-Of-File (or {@link InputStream}, here) marker. */
  39         protected boolean eof;
  40
  41         private boolean closed;
  42         private InputStream in;
  43         private int openCounter;
  44
  45         /** array + offset of pushed-back buffers */
  46         private List<Entry<byte[], Integer>> backBuffers;
  47
  48         private long bytesRead;
  49
  50         /**
  51          * Create a new {@link BufferedInputStream2} that wraps the given
  52          * {@link InputStream}.
  53          *
  54          * @param in
  55          *            the {@link InputStream} to wrap
  56          */
  57         public BufferedInputStream(InputStream in) {
  58                 this.in = in;
  59
  60                 this.buffer = new byte[BUFFER_SIZE];
  61                 this.start = 0;
  62                 this.stop = 0;
  63                 this.backBuffers = new ArrayList<Entry<byte[], Integer>>();
  64         }
  65
  66         /**
  67          * Create a new {@link BufferedInputStream2} that wraps the given bytes
  68          * array as a data source.
  69          *
  70          * @param in
  71          *            the array to wrap, cannot be NULL
  72          */
  73         public BufferedInputStream(byte[] in) {
  74                 this(in, 0, in.length);
  75         }
  76
  77         /**
  78          * Create a new {@link BufferedInputStream2} that wraps the given bytes
  79          * array as a data source.
  80          *
  81          * @param in
  82          *            the array to wrap, cannot be NULL
  83          * @param offset
  84          *            the offset to start the reading at
  85          * @param length
  86          *            the number of bytes to take into account in the array,
  87          *            starting from the offset
  88          *
  89          * @throws NullPointerException
  90          *             if the array is NULL
  91          * @throws IndexOutOfBoundsException
  92          *             if the offset and length do not correspond to the given array
  93          */
  94         public BufferedInputStream(byte[] in, int offset, int length) {
  95                 if (in == null) {
  96                         throw new NullPointerException();
  97                 } else if (offset < 0 || length < 0 || length > in.length - offset) {
  98                         throw new IndexOutOfBoundsException();
  99                 }
 100
 101                 this.in = null;
 102
 103                 this.buffer = in;
 104                 this.start = offset;
 105                 this.stop = length;
 106                 this.backBuffers = new ArrayList<Entry<byte[], Integer>>();
 107         }
 108
 109         /**
 110          * Return this very same {@link BufferedInputStream2}, but keep a counter of
 111          * how many streams were open this way. When calling
 112          * {@link BufferedInputStream2#close()}, decrease this counter if it is not
 113          * already zero instead of actually closing the stream.
 114          * <p>
 115          * You are now responsible for it &mdash; you <b>must</b> close it.
 116          * <p>
 117          * This method allows you to use a wrapping stream around this one and still
 118          * close the wrapping stream.
 119          *
 120          * @return the same stream, but you are now responsible for closing it
 121          *
 122          * @throws IOException
 123          *             in case of I/O error or if the stream is closed
 124          */
 125         public synchronized InputStream open() throws IOException {
 126                 checkClose();
 127                 openCounter++;
 128                 return this;
 129         }
 130
 131         /**
 132          * Check if the current content (until eof) is equal to the given search
 133          * term.
 134          * <p>
 135          * Note: the search term size <b>must</b> be smaller or equal the internal
 136          * buffer size.
 137          *
 138          * @param search
 139          *            the term to search for
 140          *
 141          * @return TRUE if the content that will be read starts with it
 142          *
 143          * @throws IOException
 144          *             in case of I/O error or if the size of the search term is
 145          *             greater than the internal buffer
 146          */
 147         public boolean is(String search) throws IOException {
 148                 return is(StringUtils.getBytes(search));
 149         }
 150
 151         /**
 152          * Check if the current content (until eof) is equal to the given search
 153          * term.
 154          * <p>
 155          * Note: the search term size <b>must</b> be smaller or equal the internal
 156          * buffer size.
 157          *
 158          * @param search
 159          *            the term to search for
 160          *
 161          * @return TRUE if the content that will be read starts with it
 162          *
 163          * @throws IOException
 164          *             in case of I/O error or if the size of the search term is
 165          *             greater than the internal buffer
 166          */
 167         public boolean is(byte[] search) throws IOException {
 168                 if (startsWith(search)) {
 169                         return available() == search.length;
 170                 }
 171
 172                 return false;
 173         }
 174
 175         /**
 176          * Check if the current content (what will be read next) starts with the
 177          * given search term.
 178          * <p>
 179          * Note: the search term size <b>must</b> be smaller or equal the internal
 180          * buffer size.
 181          *
 182          * @param search
 183          *            the term to search for
 184          *
 185          * @return TRUE if the content that will be read starts with it
 186          *
 187          * @throws IOException
 188          *             in case of I/O error or if the size of the search term is
 189          *             greater than the internal buffer
 190          */
 191         public boolean startsWith(String search) throws IOException {
 192                 return startsWith(StringUtils.getBytes(search));
 193         }
 194
 195         /**
 196          * Check if the current content (what will be read next) starts with the
 197          * given search term.
 198          * <p>
 199          * An empty string will always return true (unless the stream is closed,
 200          * which would throw an {@link IOException}).
 201          * <p>
 202          * Note: the search term size <b>must</b> be smaller or equal the internal
 203          * buffer size.
 204          *
 205          * @param search
 206          *            the term to search for
 207          *
 208          * @return TRUE if the content that will be read starts with it
 209          *
 210          * @throws IOException
 211          *             in case of I/O error or if the size of the search term is
 212          *             greater than the internal buffer
 213          */
 214         public boolean startsWith(byte[] search) throws IOException {
 215                 checkClose();
 216
 217                 while (consolidatePushBack(search.length) < search.length) {
 218                         preRead();
 219                         if (start >= stop) {
 220                                 // Not enough data left to start with that
 221                                 return false;
 222                         }
 223
 224                         byte[] newBuffer = new byte[stop - start];
 225                         System.arraycopy(buffer, start, newBuffer, 0, stop - start);
 226                         pushback(newBuffer, 0);
 227                         start = stop;
 228                 }
 229
 230                 Entry<byte[], Integer> bb = backBuffers.get(backBuffers.size() - 1);
 231                 byte[] bbBuffer = bb.getKey();
 232                 int bbOffset = bb.getValue();
 233
 234                 return StreamUtils.startsWith(search, bbBuffer, bbOffset,
 235                                 bbBuffer.length);
 236         }
 237
 238         /**
 239          * The number of bytes read from the under-laying {@link InputStream}.
 240          *
 241          * @return the number of bytes
 242          */
 243         public long getBytesRead() {
 244                 return bytesRead;
 245         }
 246
 247         /**
 248          * Check if this stream is spent (no more data to read or to process).
 249          *
 250          * @return TRUE if it is
 251          *
 252          * @throws IOException
 253          *             in case of I/O error
 254          */
 255         public boolean eof() throws IOException {
 256                 if (closed) {
 257                         return true;
 258                 }
 259
 260                 preRead();
 261                 return !hasMoreData();
 262         }
 263
 264         /**
 265          * Read the whole {@link InputStream} until the end and return the number of
 266          * bytes read.
 267          *
 268          * @return the number of bytes read
 269          *
 270          * @throws IOException
 271          *             in case of I/O error
 272          */
 273         public long end() throws IOException {
 274                 long skipped = 0;
 275                 while (hasMoreData()) {
 276                         skipped += skip(buffer.length);
 277                 }
 278
 279                 return skipped;
 280         }
 281
 282         @Override
 283         public int read() throws IOException {
 284                 checkClose();
 285
 286                 preRead();
 287                 if (eof) {
 288                         return -1;
 289                 }
 290
 291                 return buffer[start++];
 292         }
 293
 294         @Override
 295         public int read(byte[] b) throws IOException {
 296                 return read(b, 0, b.length);
 297         }
 298
 299         @Override
 300         public int read(byte[] b, int boff, int blen) throws IOException {
 301                 checkClose();
 302
 303                 if (b == null) {
 304                         throw new NullPointerException();
 305                 } else if (boff < 0 || blen < 0 || blen > b.length - boff) {
 306                         throw new IndexOutOfBoundsException();
 307                 } else if (blen == 0) {
 308                         return 0;
 309                 }
 310
 311                 // Read from the pushed-back buffers if any
 312                 if (!backBuffers.isEmpty()) {
 313                         int read = 0;
 314
 315                         Entry<byte[], Integer> bb = backBuffers
 316                                         .remove(backBuffers.size() - 1);
 317                         byte[] bbBuffer = bb.getKey();
 318                         int bbOffset = bb.getValue();
 319                         int bbSize = bbBuffer.length - bbOffset;
 320
 321                         if (bbSize > blen) {
 322                                 read = blen;
 323                                 System.arraycopy(bbBuffer, bbOffset, b, boff, read);
 324                                 pushback(bbBuffer, bbOffset + read);
 325                         } else {
 326                                 read = bbSize;
 327                                 System.arraycopy(bbBuffer, bbOffset, b, boff, read);
 328                         }
 329
 330                         return read;
 331                 }
 332
 333                 int done = 0;
 334                 while (hasMoreData() && done < blen) {
 335                         preRead();
 336                         if (hasMoreData()) {
 337                                 int now = Math.min(blen - done, stop - start);
 338                                 if (now > 0) {
 339                                         System.arraycopy(buffer, start, b, boff + done, now);
 340                                         start += now;
 341                                         done += now;
 342                                 }
 343                         }
 344                 }
 345
 346                 return done > 0 ? done : -1;
 347         }
 348
 349         @Override
 350         public long skip(long n) throws IOException {
 351                 if (n <= 0) {
 352                         return 0;
 353                 }
 354
 355                 long skipped = 0;
 356                 while (!backBuffers.isEmpty() && n > 0) {
 357                         Entry<byte[], Integer> bb = backBuffers
 358                                         .remove(backBuffers.size() - 1);
 359                         byte[] bbBuffer = bb.getKey();
 360                         int bbOffset = bb.getValue();
 361                         int bbSize = bbBuffer.length - bbOffset;
 362
 363                         int localSkip = 0;
 364                         localSkip = (int) Math.min(n, bbSize);
 365
 366                         n -= localSkip;
 367                         bbSize -= localSkip;
 368
 369                         if (bbSize > 0) {
 370                                 pushback(bbBuffer, bbOffset + localSkip);
 371                         }
 372                 }
 373                 while (hasMoreData() && n > 0) {
 374                         preRead();
 375
 376                         long inBuffer = Math.min(n, available());
 377                         start += inBuffer;
 378                         n -= inBuffer;
 379                         skipped += inBuffer;
 380                 }
 381
 382                 return skipped;
 383         }
 384
 385         @Override
 386         public int available() {
 387                 if (closed) {
 388                         return 0;
 389                 }
 390
 391                 int avail = 0;
 392                 for (Entry<byte[], Integer> entry : backBuffers) {
 393                         avail += entry.getKey().length - entry.getValue();
 394                 }
 395
 396                 return avail + Math.max(0, stop - start);
 397         }
 398
 399         /**
 400          * Closes this stream and releases any system resources associated with the
 401          * stream.
 402          * <p>
 403          * Including the under-laying {@link InputStream}.
 404          * <p>
 405          * <b>Note:</b> if you called the {@link BufferedInputStream2#open()} method
 406          * prior to this one, it will just decrease the internal count of how many
 407          * open streams it held and do nothing else. The stream will actually be
 408          * closed when you have called {@link BufferedInputStream2#close()} once
 409          * more than {@link BufferedInputStream2#open()}.
 410          *
 411          * @exception IOException
 412          *                in case of I/O error
 413          */
 414         @Override
 415         public synchronized void close() throws IOException {
 416                 close(true);
 417         }
 418
 419         /**
 420          * Closes this stream and releases any system resources associated with the
 421          * stream.
 422          * <p>
 423          * Including the under-laying {@link InputStream} if
 424          * <tt>incudingSubStream</tt> is true.
 425          * <p>
 426          * You can call this method multiple times, it will not cause an
 427          * {@link IOException} for subsequent calls.
 428          * <p>
 429          * <b>Note:</b> if you called the {@link BufferedInputStream2#open()} method
 430          * prior to this one, it will just decrease the internal count of how many
 431          * open streams it held and do nothing else. The stream will actually be
 432          * closed when you have called {@link BufferedInputStream2#close()} once
 433          * more than {@link BufferedInputStream2#open()}.
 434          *
 435          * @param includingSubStream
 436          *            also close the under-laying stream
 437          *
 438          * @exception IOException
 439          *                in case of I/O error
 440          */
 441         public synchronized void close(boolean includingSubStream)
 442                         throws IOException {
 443                 if (!closed) {
 444                         if (openCounter > 0) {
 445                                 openCounter--;
 446                         } else {
 447                                 closed = true;
 448                                 if (includingSubStream && in != null) {
 449                                         in.close();
 450                                 }
 451                         }
 452                 }
 453         }
 454
 455         /**
 456          * Consolidate the push-back buffers so the last one is at least the given
 457          * size, if possible.
 458          * <p>
 459          * If there is not enough data in the push-back buffers, they will all be
 460          * consolidated.
 461          *
 462          * @param size
 463          *            the minimum size of the consolidated buffer, or -1 to force
 464          *            the consolidation of all push-back buffers
 465          *
 466          * @return the size of the last, consolidated buffer; can be less than the
 467          *         requested size if not enough data
 468          */
 469         protected int consolidatePushBack(int size) {
 470                 int bbIndex = -1;
 471                 int bbUpToSize = 0;
 472                 for (Entry<byte[], Integer> entry : backBuffers) {
 473                         bbIndex++;
 474                         bbUpToSize += entry.getKey().length - entry.getValue();
 475
 476                         if (size >= 0 && bbUpToSize >= size) {
 477                                 break;
 478                         }
 479                 }
 480
 481                 // Index 0 means "the last buffer is already big enough"
 482                 if (bbIndex > 0) {
 483                         byte[] consolidatedBuffer = new byte[bbUpToSize];
 484                         int consolidatedPos = 0;
 485                         for (int i = 0; i <= bbIndex; i++) {
 486                                 Entry<byte[], Integer> bb = backBuffers
 487                                                 .remove(backBuffers.size() - 1);
 488                                 byte[] bbBuffer = bb.getKey();
 489                                 int bbOffset = bb.getValue();
 490                                 int bbSize = bbBuffer.length - bbOffset;
 491                                 System.arraycopy(bbBuffer, bbOffset, consolidatedBuffer,
 492                                                 consolidatedPos, bbSize);
 493                         }
 494
 495                         pushback(consolidatedBuffer, 0);
 496                 }
 497
 498                 return bbUpToSize;
 499         }
 500
 501         /**
 502          * Check if we still have some data in the buffer and, if not, fetch some.
 503          *
 504          * @return TRUE if we fetched some data, FALSE if there are still some in
 505          *         the buffer
 506          *
 507          * @throws IOException
 508          *             in case of I/O error
 509          */
 510         protected boolean preRead() throws IOException {
 511                 boolean hasRead = false;
 512                 if (in != null && !eof && start >= stop) {
 513                         start = 0;
 514                         stop = read(in, buffer, 0, buffer.length);
 515                         if (stop > 0) {
 516                                 bytesRead += stop;
 517                         }
 518
 519                         hasRead = true;
 520                 }
 521
 522                 if (start >= stop) {
 523                         eof = true;
 524                 }
 525
 526                 return hasRead;
 527         }
 528
 529         /**
 530          * Push back some data that will be read again at the next read call.
 531          *
 532          * @param buffer
 533          *            the buffer to push back
 534          * @param offset
 535          *            the offset at which to start reading in the buffer
 536          */
 537         protected void pushback(byte[] buffer, int offset) {
 538                 backBuffers.add(
 539                                 new AbstractMap.SimpleEntry<byte[], Integer>(buffer, offset));
 540         }
 541
 542         /**
 543          * Read the under-laying stream into the given local buffer.
 544          *
 545          * @param in
 546          *            the under-laying {@link InputStream}
 547          * @param buffer
 548          *            the buffer we use in this {@link BufferedInputStream2}
 549          * @param off
 550          *            the offset
 551          * @param len
 552          *            the length in bytes
 553          *
 554          * @return the number of bytes read
 555          *
 556          * @throws IOException
 557          *             in case of I/O error
 558          */
 559         protected int read(InputStream in, byte[] buffer, int off, int len)
 560                         throws IOException {
 561                 return in.read(buffer, off, len);
 562         }
 563
 564         /**
 565          * We have more data available in the buffer <b>or</b> we can, maybe, fetch
 566          * more.
 567          *
 568          * @return TRUE if it is the case, FALSE if not
 569          */
 570         protected boolean hasMoreData() {
 571                 if (closed) {
 572                         return false;
 573                 }
 574
 575                 return !backBuffers.isEmpty() || (start < stop) || !eof;
 576         }
 577
 578         /**
 579          * Check that the stream was not closed, and throw an {@link IOException} if
 580          * it was.
 581          *
 582          * @throws IOException
 583          *             if it was closed
 584          */
 585         protected void checkClose() throws IOException {
 586                 if (closed) {
 587                         throw new IOException(
 588                                         "This BufferedInputStream was closed, you cannot use it anymore.");
 589                 }
 590         }
 591 }