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          * <p>
  89          * This is can be a blocking call when data need to be fetched.
  90          *
  91          * @return TRUE if we unblocked the next sub-stream, FALSE if not
  92          *
  93          * @throws IOException
  94          *             in case of I/O error or if the stream is closed
  95          */
  96         public boolean next() throws IOException {
  97                 return next(false);
  98         }
  99
 100         /**
 101          * Unblock the next sub-stream as would have done
 102          * {@link NextableInputStream#next()}, but disable the sub-stream systems.
 103          * <p>
 104          * That is, the next stream, if any, will be the last one and will not be
 105          * subject to the {@link NextableInputStreamStep}.
 106          * <p>
 107          * This is can be a blocking call when data need to be fetched.
 108          *
 109          * @return TRUE if we unblocked the next sub-stream, FALSE if not
 110          *
 111          * @throws IOException
 112          *             in case of I/O error or if the stream is closed
 113          */
 114         public boolean nextAll() throws IOException {
 115                 return next(true);
 116         }
 117
 118         /**
 119          * Check if this stream is totally spent (no more data to read or to
 120          * process).
 121          * <p>
 122          * Note: when the stream is divided into sub-streams, each sub-stream will
 123          * report it is eof when emptied.
 124          *
 125          * @return TRUE if it is
 126          *
 127          * @throws IOException
 128          *             in case of I/O error
 129          */
 130         @Override
 131         public boolean eof() throws IOException {
 132                 return super.eof();
 133         }
 134
 135         /**
 136          * Check if we still have some data in the buffer and, if not, fetch some.
 137          *
 138          * @return TRUE if we fetched some data, FALSE if there are still some in
 139          *         the buffer
 140          *
 141          * @throws IOException
 142          *             in case of I/O error
 143          */
 144         @Override
 145         protected boolean preRead() throws IOException {
 146                 if (!stopped) {
 147                         boolean bufferChanged = super.preRead();
 148                         checkBuffer(bufferChanged);
 149                         return bufferChanged;
 150                 }
 151
 152                 if (start >= stop) {
 153                         eof = true;
 154                 }
 155
 156                 return false;
 157         }
 158
 159         /**
 160          * We have more data available in the buffer or we can fetch more.
 161          *
 162          * @return TRUE if it is the case, FALSE if not
 163          */
 164         @Override
 165         protected boolean hasMoreData() {
 166                 return started && super.hasMoreData();
 167         }
 168
 169         /**
 170          * Check that the buffer didn't overshot to the next item, and fix
 171          * {@link NextableInputStream#stop} if needed.
 172          * <p>
 173          * If {@link NextableInputStream#stop} is fixed,
 174          * {@link NextableInputStream#eof} and {@link NextableInputStream#stopped}
 175          * are set to TRUE.
 176          *
 177          * @param newBuffer
 178          *            we changed the buffer, we need to clear some information in
 179          *            the {@link NextableInputStreamStep}
 180          */
 181         private void checkBuffer(boolean newBuffer) {
 182                 if (step != null && stop >= 0) {
 183                         if (newBuffer) {
 184                                 step.clearBuffer();
 185                         }
 186
 187                         int stopAt = step.stop(buffer, start, stop, eof);
 188                         if (stopAt >= 0) {
 189                                 stop = stopAt;
 190                                 eof = true;
 191                                 stopped = true;
 192                         }
 193                 }
 194         }
 195
 196         /**
 197          * The implementation of {@link NextableInputStream#next()} and
 198          * {@link NextableInputStream#nextAll()}.
 199          * <p>
 200          * This is can be a blocking call when data need to be fetched.
 201          *
 202          * @param all
 203          *            TRUE for {@link NextableInputStream#nextAll()}, FALSE for
 204          *            {@link NextableInputStream#next()}
 205          *
 206          * @return TRUE if we unblocked the next sub-stream, FALSE if not
 207          *
 208          * @throws IOException
 209          *             in case of I/O error or if the stream is closed
 210          */
 211         private boolean next(boolean all) throws IOException {
 212                 checkClose();
 213
 214                 if (!started) {
 215                         // First call before being allowed to read
 216                         started = true;
 217
 218                         if (all) {
 219                                 step = null;
 220                         }
 221                 }
 222
 223                 if (step != null && !hasMoreData() && stopped) {
 224                         stop = step.getResumeLen();
 225                         start += step.getResumeSkip();
 226                         eof = step.getResumeEof();
 227                         stopped = false;
 228
 229                         if (all) {
 230                                 step = null;
 231                         }
 232
 233                         checkBuffer(false);
 234                 }
 235
 236                 // consider that if EOF, there is no next
 237                 if (start >= stop) {
 238                         // Make sure, block if necessary
 239                         preRead();
 240
 241                         return hasMoreData();
 242                 }
 243
 244                 return true;
 245         }
 246
 247         public String DEBUG() {
 248                 String rep = String.format(
 249                                 "Nextable %s: %d -> %d [eof: %s] [more data: %s]",
 250                                 (stopped ? "stopped" : "running"), start, stop, "" + eof, ""
 251                                                 + hasMoreData());
 252
 253                 return rep;
 254         }
 255 }