// We effectively turn the stream into a non-blocking stream.

// There is the addition isEOF() to check whether an actual

// stream end has been reached (true) or whether we are just

// waiting for more data (false).

sclass ByteArraysPartialInputStream extends InputStreamPlusReadFully {

  sclass State {

    long absolutePosition;

    int ofs;

    MinimalChain<byte[]> chain;

    *(long *absolutePosition, int *ofs, MinimalChain<byte[]> *chain) {}

    byte[] data() { ret chain!; }

    int nBytes() { ret l(data()); }

    bool exhausted() { ret ofs >= nBytes(); }

    toString {

      ret withIdentity(this, "State" + bracketed(

        renderVars(+ofs, +nBytes(), next := chain.next))); }

  }

  bool debug, eof;

  MinimalChain<byte[]> last = new MinimalChain(null);

  State state = new(0, 0, last);

  long remaining;

  State mark;

  @Override public synchronized void readFully(byte[] buf, int off, int n) throws IOException {

    if (debug) printFunctionCall readFully(buf, off, n);

    if (n == 0) ret;

    if (n > available())

      throw new EOFException();

    assertEquals(n, read(buf, off, n));

  }

  @Override public synchronized int read(byte[] buf, int off, int len) throws IOException {

    if (debug) printFunctionCall read(buf, off, len);

    len = min(len, available());

    if (len == 0) ret -1;

    int n = 0;

    while (state.chain != last) {

      int nChunk = state.nBytes();

      int inChunk = min(len, nChunk-state.ofs);

      int copiedFrom = state.ofs, nBytes = state.nBytes();

      arraycopy(state.data(), copiedFrom, buf, off, inChunk);

      n += inChunk;

      off += inChunk;

      len -= inChunk;

      state.ofs += inChunk;

      state.absolutePosition += inChunk;

      remaining -= inChunk;

      if (debug) printVars(nBytes := state.nBytes(), +copiedFrom, +nChunk, +inChunk, +n, +off, +len);

      if (state.exhausted())

        state = new State(state.absolutePosition, 0, state.chain.next);

      else

        break;

    }

    ret n;

  }

  @Override public synchronized int available() { ret clampToInt(remaining); }

  @Override public synchronized int read() throws IOException {

    if (debug) printFunctionCall read();

    while ping (true) {

      if (state.ofs < l(state.chain!)) {

        ++state.absolutePosition;

        --remaining;

        ret ubyteToInt(state.chain![state.ofs++]);

      }

      if (state.chain.next == null)

        ret noElement();

      state = new State(0, 0, state.chain.next);

    }

  }

  static int noElement() { ret -1; }

  public synchronized void add aka write(byte[] buffer) {

    if (empty(buffer)) ret;

    remaining += l(buffer);

    last.setNext(new MinimalChain(null));

    last.setElement(buffer);

    last = last.next;

  }

  public synchronized void addEOF() {

    eof = true;

  }

  public synchronized bool isEOF() {

    ret state.chain == last && eof;

  }

  @Override public synchronized bool markSupported() { true; }

  @Override public synchronized void mark(int readLimit) {

    if (debug) printFunctionCall mark(readLimit);

    // clone state

    mark = new State(state.absolutePosition, state.ofs, state.chain);

    ifdef ByteArraysPartialInputStream_debug

      printVars(+mark);

    endifdef

 }

  @Override public synchronized void reset() throws IOException {

    if (debug) printFunctionCall reset();

    if (mark == null) fail("Reset without mark");

    remaining += state.absolutePosition-mark.absolutePosition;

    state = mark;

    mark = null;

  }

  synchronized S stats() { ret toStringWithFields(this); }

}