src/be/nikiroo/utils/streams/NextableInputStream.java

   1 package be.nikiroo.utils.streams;
   2
   3 import java.io.IOException;
   4 import java.io.InputStream;
   5
   6 /**
   7  * This {@link InputStream} can be separated into sub-streams (you can process
   8  * it as a normal {@link InputStream} but, when it is spent, you can call
   9  * {@link NextableInputStream#next()} on it to unlock new data).
  10  * <p>
  11  * The separation in sub-streams is done via {@link NextableInputStreamStep}.
  12  *
  13  * @author niki
  14  */
  15 public class NextableInputStream extends BufferedInputStream {
  16         private NextableInputStreamStep step;
  17         private boolean started;
  18         private boolean stopped;
  19
  20         /**
  21          * Create a new {@link NextableInputStream} that wraps the given
  22          * {@link InputStream}.
  23          *
  24          * @param in
  25          *            the {@link InputStream} to wrap
  26          * @param step
  27          *            how to separate it into sub-streams (can be NULL, but in that
  28          *            case it will behave as a normal {@link InputStream})
  29          */
  30         public NextableInputStream(InputStream in, NextableInputStreamStep step) {
  31                 super(in);
  32                 this.step = step;
  33         }
  34
  35         /**
  36          * Create a new {@link NextableInputStream} that wraps the given bytes array
  37          * as a data source.
  38          *
  39          * @param in
  40          *            the array to wrap, cannot be NULL
  41          * @param step
  42          *            how to separate it into sub-streams (can be NULL, but in that
  43          *            case it will behave as a normal {@link InputStream})
  44          */
  45         public NextableInputStream(byte[] in, NextableInputStreamStep step) {
  46                 this(in, step, 0, in.length);
  47         }
  48
  49         /**
  50          * Create a new {@link NextableInputStream} that wraps the given bytes array
  51          * as a data source.
  52          *
  53          * @param in
  54          *            the array to wrap, cannot be NULL
  55          * @param step
  56          *            how to separate it into sub-streams (can be NULL, but in that
  57          *            case it will behave as a normal {@link InputStream})
  58          * @param offset
  59          *            the offset to start the reading at
  60          * @param length
  61          *            the number of bytes to take into account in the array,
  62          *            starting from the offset
  63          *
  64          * @throws NullPointerException
  65          *             if the array is NULL
  66          * @throws IndexOutOfBoundsException
  67          *             if the offset and length do not correspond to the given array
  68          */
  69         public NextableInputStream(byte[] in, NextableInputStreamStep step,
  70                         int offset, int length) {
  71                 super(in, offset, length);
  72                 this.step = step;
  73                 checkBuffer(true);
  74         }
  75
  76         /**
  77          * Unblock the processing of the next sub-stream.
  78          * <p>
  79          * It can only be called when the "current" stream is spent (i.e., you must
  80          * first process the stream until it is spent).
  81          * <p>
  82          * We consider that when the under-laying {@link InputStream} is also spent,
  83          * we cannot have a next sub-stream (it will thus return FALSE).
  84          * <p>
  85          * {@link IOException}s can happen when we have no data available in the
  86          * buffer; in that case, we fetch more data to know if we can have a next
  87          * sub-stream or not.
  88          *
  89          * @return TRUE if we unblocked the next sub-stream, FALSE if not
  90          *
  91          * @throws IOException
  92          *             in case of I/O error or if the stream is closed
  93          */
  94         public boolean next() throws IOException {
  95                 return next(false);
  96         }
  97
  98         /**
  99          * Unblock the next sub-stream as would have done
 100          * {@link NextableInputStream#next()}, but disable the sub-stream systems.
 101          * <p>
 102          * That is, the next stream, if any, will be the last one and will not be
 103          * subject to the {@link NextableInputStreamStep}.
 104          *
 105          * @return TRUE if we unblocked the next sub-stream, FALSE if not
 106          *
 107          * @throws IOException
 108          *             in case of I/O error or if the stream is closed
 109          */
 110         public boolean nextAll() throws IOException {
 111                 return next(true);
 112         }
 113
 114         /**
 115          * Check if this stream is totally spent (no more data to read or to
 116          * process).
 117          * <p>
 118          * Note: when the stream is divided into sub-streams, each sub-stream will
 119          * report it is eof when emptied.
 120          *
 121          * @return TRUE if it is
 122          *
 123          * @throws IOException
 124          *             in case of I/O error
 125          */
 126         @Override
 127         public boolean eof() throws IOException {
 128                 return super.eof();
 129         }
 130
 131         /**
 132          * Check if we still have some data in the buffer and, if not, fetch some.
 133          *
 134          * @return TRUE if we fetched some data, FALSE if there are still some in
 135          *         the buffer
 136          *
 137          * @throws IOException
 138          *             in case of I/O error
 139          */
 140         @Override
 141         protected boolean preRead() throws IOException {
 142                 if (!stopped) {
 143                         boolean bufferChanged = super.preRead();
 144                         checkBuffer(bufferChanged);
 145                         return bufferChanged;
 146                 }
 147
 148                 if (start >= stop) {
 149                         eof = true;
 150                 }
 151
 152                 return false;
 153         }
 154
 155         /**
 156          * We have more data available in the buffer or we can fetch more.
 157          *
 158          * @return TRUE if it is the case, FALSE if not
 159          */
 160         @Override
 161         protected boolean hasMoreData() {
 162                 return started && super.hasMoreData();
 163         }
 164
 165         /**
 166          * Check that the buffer didn't overshot to the next item, and fix
 167          * {@link NextableInputStream#stop} if needed.
 168          * <p>
 169          * If {@link NextableInputStream#stop} is fixed,
 170          * {@link NextableInputStream#eof} and {@link NextableInputStream#stopped}
 171          * are set to TRUE.
 172          *
 173          * @param newBuffer
 174          *            we changed the buffer, we need to clear some information in
 175          *            the {@link NextableInputStreamStep}
 176          */
 177         private void checkBuffer(boolean newBuffer) {
 178                 if (step != null && stop >= 0) {
 179                         if (newBuffer) {
 180                                 step.clearBuffer();
 181                         }
 182
 183                         int stopAt = step.stop(buffer, start, stop, eof);
 184                         if (stopAt >= 0) {
 185                                 stop = stopAt;
 186                                 eof = true;
 187                                 stopped = true;
 188                         }
 189                 }
 190         }
 191
 192         /**
 193          * The implementation of {@link NextableInputStream#next()} and
 194          * {@link NextableInputStream#nextAll()}.
 195          *
 196          * @param all
 197          *            TRUE for {@link NextableInputStream#nextAll()}, FALSE for
 198          *            {@link NextableInputStream#next()}
 199          *
 200          * @return TRUE if we unblocked the next sub-stream, FALSE if not
 201          *
 202          * @throws IOException
 203          *             in case of I/O error or if the stream is closed
 204          */
 205         private boolean next(boolean all) throws IOException {
 206                 checkClose();
 207
 208                 if (!started) {
 209                         // First call before being allowed to read
 210                         started = true;
 211
 212                         if (all) {
 213                                 step = null;
 214                         }
 215
 216                         return true;
 217                 }
 218
 219                 if (step != null && !hasMoreData() && stopped) {
 220                         stop = step.getResumeLen();
 221                         start += step.getResumeSkip();
 222                         eof = step.getResumeEof();
 223                         stopped = false;
 224
 225                         if (all) {
 226                                 step = null;
 227                         }
 228
 229                         checkBuffer(false);
 230
 231                         // consider that if EOF, there is no next
 232                         return hasMoreData();
 233                 }
 234
 235                 return false;
 236         }
 237 }