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