<commit_before>part of dartz_streaming;

/*
  WARNING: Experimental, incomplete and somewhat buggy! Not fully stack safe yet! Will most likely change substantially over time.
  Design similar to the "Process" construct from chapter 15 of "Functional Programming in Scala" by <NAME> and <NAME>.
 */

typedef Conveyor<F, Unit> SinkF<F, O>(O o);
typedef Conveyor<F, O> ChannelF<F, I, O>(I i);

abstract class Conveyor<F, O> extends FunctorOps<Conveyor/*<F, dynamic>*/, O> with ApplicativeOps<Conveyor/*<F, dynamic>*/, O>, ApplicativePlusOps<Conveyor/*<F, dynamic>*/, O>, MonadOps<Conveyor/*<F, dynamic>*/, O>, MonadPlusOps<Conveyor/*<F, dynamic>*/, O> {

  /*=A*/ interpret/*<A>*/(/*=A*/ ifProduce(O head, Conveyor<F, O> tail), /*=A*/ ifConsume(F req, Function1<Either<Object, dynamic>, Conveyor<F, O>> recv), /*=A*/ ifHalt(Object err));

  static Conveyor/*<F, O>*/ produce/*<F, O>*/(/*=O*/ head, [Conveyor/*<F, O>*/ tail]) => new _Produce(head, tail ?? halt(End));
  static Conveyor/*<F, O>*/ consume/*<F, A, O>*/(/*=F*/ req, Function1<Either<Object, dynamic/*=A*/>, Conveyor/*<F, O>*/> recv) => new _Consume/*<F, A, O>*/(req, recv);
  static Conveyor/*<F, O>*/ halt/*<F, O>*/([Object err]) => new _Halt(err ?? End);

  static final End = new _End();
  static final Kill = new _Kill();

  Conveyor<F, dynamic/*=B*/> pure/*<B>*/(/*=B*/ b) => produce(b);

  Conveyor<F, dynamic/*=O2*/> map/*<O2>*/(/*=O2*/ f(O o)) =>
      interpret((h, t) => tryOrDie(() => produce(f(h), t.map/*<O2>*/(f))),
          (req, recv) => consume(req, (ea) => recv(ea).map/*<O2>*/(f)),
          halt);

  Conveyor<F, O> lazyPlus(Conveyor<F, O> p()) =>
      onHaltEnd(() => Try(p));

  Conveyor<F, O> empty() => halt(End);

  Conveyor<F, O> plus(Conveyor<F, O> p) =>
      onHaltEnd(() => p);

  Conveyor<F, O> operator +(Conveyor<F, O> p) => plus(p);

  Conveyor<F, O> onComplete(Conveyor<F, O> p()) =>
      onHalt((err) => err == End ? p().asFinalizer() : p().asFinalizer().plus(halt(err)));

  Conveyor<F, O> asFinalizer() =>
      interpret((h, t) => produce(h, t.asFinalizer()),
          (req, recv) => consume(req, (ea) => ea == left(Kill) ? asFinalizer() : recv(ea)),
          halt);

  Conveyor<F, O> onHalt(Conveyor<F, O> f(Object err)) =>
      interpret((h, t) => produce(h, t.onHalt(f)),
          (req, recv) => consume(req, (ea) => recv(ea).onHalt(f)),
          (err) => Try(() => f(err)));

  Conveyor<F, O> onHaltEnd(Conveyor<F, O> f()) =>
      onHalt((err) => err == End ? f() : halt(err));

  Conveyor<F, dynamic/*=O2*/> flatMap/*<O2>*/(Conveyor<F, dynamic/*=O2*/> f(O o)) =>
      interpret((h, t) => tryOrDie(() => (f(h).onHalt((err) => err == End ? halt(End) : kill/*<O2>*/().plus(halt(err)))).lazyPlus(() => t.flatMap /*<O2>*/(f))),
          (req, recv) => consume(req, (ea) => recv(ea).flatMap/*<O2>*/(f)),
          halt);

  Conveyor<F, dynamic/*=O2*/> bind/*<O2>*/(Conveyor<F, dynamic/*=O2*/> f(O o)) => flatMap(f);

  Conveyor<F, O> repeat() => lazyPlus(repeat);

  Conveyor<F, O> repeatUntilExhausted() =>
      interpret((h, t) => produce(h, t.lazyPlus(repeatUntilExhausted)),
          (req, recv) => consume(req, (ea) => ea.fold((l) => recv(left(l)), (r) => recv(right(r)).lazyPlus(repeatUntilExhausted))),
          halt);

  Conveyor<F, O> repeatNonEmpty() {
    final cycle = this.map(some).lazyPlus(() => produce(none())).repeat();
    final sentinel = tuple2(some(none()), none());
    final trimmed = cycle.pipe(Pipe.window2()).takeWhile((pair) => pair != sentinel);
    return trimmed.map((t) => t.value2).flatMap((o) => o.fold(() => halt(End), (v) => produce(v)));
  }

  F/*=FLA*/ runLog/*<FLA extends F>*/(MonadCatch<F> monadCatch) {
    F go(Conveyor<F, O> cur, IList<O> acc) =>
        cur.interpret((h, t) => go(t, cons(h, acc)),
            (req, recv) => monadCatch.bind(monadCatch.attempt(req), (Either<Object, dynamic> e) => go(Try(() => recv(e)), acc)),
            (err) => err == End ? monadCatch.pure(acc.reverse()) : monadCatch.fail(err));
    return go(this, nil()) as dynamic/*=FLA*/;
  }

  Conveyor<F, dynamic/*=O2*/> drain/*<O2>*/() =>
      interpret((h, t) => t.drain/*<O2>*/(),
          (req, recv) => consume(req, (ea) => recv(ea).drain/*<O2>*/()),
          halt);

  Conveyor<F, dynamic/*=O2*/> kill/*<O2>*/() =>
      interpret((h, t) => t.kill/*<O2>*/(),
          (req, recv) => recv(left(Kill)).drain/*<O2>*/().onHalt((e) => e == Kill ? halt(End) : halt(e)),
          halt);


  Conveyor<F, dynamic/*=O2*/> pipe/*<O2>*/(Conveyor<From<O>, dynamic/*=O2*/> c2) =>
      c2.interpret((h, t) => produce(h, pipe /*<O2>*/(t)),
          (req, recv) =>
              this.interpret((h, t) => t.pipe/*<O2>*/(Try(() => recv(right(h)))),
                  (req0, recv0) => consume/*<F, O, O2>*/(req0, (ea) => recv0(ea).pipe /*<O2>*/(c2)),
                  (err) => halt/*<F, O>*/(err).pipe/*<O2>*/(recv(left(err)))),
          (err) => kill/*<O2>*/().onHalt((err2) => halt/*<F, O2>*/(err).plus(halt(err2))));

  Conveyor<F, dynamic> operator |(Conveyor<From<O>, dynamic> c2) => pipe(c2);

  Conveyor<F, O> lastOr(O o) =>
      interpret((h, t) => t.lastOr(h),
          (req, recv) => consume(req, (ea) => recv(ea).lastOr(o)),
          (err) => err == End ? produce(o) : halt(err));

  static Conveyor/*<F, O>*/ Try/*<F, O>*/(Conveyor/*<F, O>*/ p()) {
    try {
      return p();
    } catch (err) {
      return halt(err);
    }
  }

  Conveyor/*<F, O2>*/ tryOrDie/*<O2>*/(Conveyor/*<F, O2>*/ p()) {
    try {
      return p();
    } catch (err) {
      return kill/*<O2>*/().onHalt((err2) => halt/*<F, O2>*/(err).plus(halt(err2)));
    }
  }

  Conveyor<F, O> take(int n) => pipe(Pipe.take(n));

  Conveyor<F, O> takeWhile(bool f(O o)) => pipe(Pipe.takeWhile(f));

  Conveyor<F, O> drop(int n) => pipe(Pipe.drop(n));

  Conveyor<F, O> dropWhile(bool f(O o)) => pipe(Pipe.dropWhile(f));

  Conveyor<F, O> filter(bool f(O o)) => pipe(Pipe.filter(f));

  Conveyor<F, dynamic/*=O2*/> fold/*<O2>*/(/*=O2*/ z, Function2/*<O2, O, O2>*/ f) => pipe(Pipe.scan(z, f)).lastOr(z);

  Conveyor<F, O> concatenate(Monoid<O> monoid) => pipe(Pipe.scan(monoid.zero(), monoid.append)).lastOr(monoid.zero());
<commit_msg>Added foldWhile to Conveyor<commit_after>part of dartz_streaming;

/*
  WARNING: Experimental, incomplete and somewhat buggy! Not fully stack safe yet! Will most likely change substantially over time.
  Design similar to the "Process" construct from chapter 15 of "Functional Programming in Scala" by <NAME> and <NAME>.
 */

typedef Conveyor<F, Unit> SinkF<F, O>(O o);
typedef Conveyor<F, O> ChannelF<F, I, O>(I i);

abstract class Conveyor<F, O> extends FunctorOps<Conveyor/*<F, dynamic>*/, O> with ApplicativeOps<Conveyor/*<F, dynamic>*/, O>, ApplicativePlusOps<Conveyor/*<F, dynamic>*/, O>, MonadOps<Conveyor/*<F, dynamic>*/, O>, MonadPlusOps<Conveyor/*<F, dynamic>*/, O> {

  /*=A*/ interpret/*<A>*/(/*=A*/ ifProduce(O head, Conveyor<F, O> tail), /*=A*/ ifConsume(F req, Function1<Either<Object, dynamic>, Conveyor<F, O>> recv), /*=A*/ ifHalt(Object err));

  static Conveyor/*<F, O>*/ produce/*<F, O>*/(/*=O*/ head, [Conveyor/*<F, O>*/ tail]) => new _Produce(head, tail ?? halt(End));
  static Conveyor/*<F, O>*/ consume/*<F, A, O>*/(/*=F*/ req, Function1<Either<Object, dynamic/*=A*/>, Conveyor/*<F, O>*/> recv) => new _Consume/*<F, A, O>*/(req, recv);
  static Conveyor/*<F, O>*/ halt/*<F, O>*/([Object err]) => new _Halt(err ?? End);

  static final End = new _End();
  static final Kill = new _Kill();

  Conveyor<F, dynamic/*=B*/> pure/*<B>*/(/*=B*/ b) => produce(b);

  Conveyor<F, dynamic/*=O2*/> map/*<O2>*/(/*=O2*/ f(O o)) =>
      interpret((h, t) => tryOrDie(() => produce(f(h), t.map/*<O2>*/(f))),
          (req, recv) => consume(req, (ea) => recv(ea).map/*<O2>*/(f)),
          halt);

  Conveyor<F, O> lazyPlus(Conveyor<F, O> p()) =>
      onHaltEnd(() => Try(p));

  Conveyor<F, O> empty() => halt(End);

  Conveyor<F, O> plus(Conveyor<F, O> p) =>
      onHaltEnd(() => p);

  Conveyor<F, O> operator +(Conveyor<F, O> p) => plus(p);

  Conveyor<F, O> onComplete(Conveyor<F, O> p()) =>
      onHalt((err) => err == End ? p().asFinalizer() : p().asFinalizer().plus(halt(err)));

  Conveyor<F, O> asFinalizer() =>
      interpret((h, t) => produce(h, t.asFinalizer()),
          (req, recv) => consume(req, (ea) => ea == left(Kill) ? asFinalizer() : recv(ea)),
          halt);

  Conveyor<F, O> onHalt(Conveyor<F, O> f(Object err)) =>
      interpret((h, t) => produce(h, t.onHalt(f)),
          (req, recv) => consume(req, (ea) => recv(ea).onHalt(f)),
          (err) => Try(() => f(err)));

  Conveyor<F, O> onHaltEnd(Conveyor<F, O> f()) =>
      onHalt((err) => err == End ? f() : halt(err));

  Conveyor<F, dynamic/*=O2*/> flatMap/*<O2>*/(Conveyor<F, dynamic/*=O2*/> f(O o)) =>
      interpret((h, t) => tryOrDie(() => (f(h).onHalt((err) => err == End ? halt(End) : kill/*<O2>*/().plus(halt(err)))).lazyPlus(() => t.flatMap /*<O2>*/(f))),
          (req, recv) => consume(req, (ea) => recv(ea).flatMap/*<O2>*/(f)),
          halt);

  Conveyor<F, dynamic/*=O2*/> bind/*<O2>*/(Conveyor<F, dynamic/*=O2*/> f(O o)) => flatMap(f);

  Conveyor<F, O> repeat() => lazyPlus(repeat);

  Conveyor<F, O> repeatUntilExhausted() =>
      interpret((h, t) => produce(h, t.lazyPlus(repeatUntilExhausted)),
          (req, recv) => consume(req, (ea) => ea.fold((l) => recv(left(l)), (r) => recv(right(r)).lazyPlus(repeatUntilExhausted))),
          halt);

  Conveyor<F, O> repeatNonEmpty() {
    final cycle = this.map(some).lazyPlus(() => produce(none())).repeat();
    final sentinel = tuple2(some(none()), none());
    final trimmed = cycle.pipe(Pipe.window2()).takeWhile((pair) => pair != sentinel);
    return trimmed.map((t) => t.value2).flatMap((o) => o.fold(() => halt(End), (v) => produce(v)));
  }

  F/*=FLA*/ runLog/*<FLA extends F>*/(MonadCatch<F> monadCatch) {
    F go(Conveyor<F, O> cur, IList<O> acc) =>
        cur.interpret((h, t) => go(t, cons(h, acc)),
            (req, recv) => monadCatch.bind(monadCatch.attempt(req), (Either<Object, dynamic> e) => go(Try(() => recv(e)), acc)),
            (err) => err == End ? monadCatch.pure(acc.reverse()) : monadCatch.fail(err));
    return go(this, nil()) as dynamic/*=FLA*/;
  }

  Conveyor<F, dynamic/*=O2*/> drain/*<O2>*/() =>
      interpret((h, t) => t.drain/*<O2>*/(),
          (req, recv) => consume(req, (ea) => recv(ea).drain/*<O2>*/()),
          halt);

  Conveyor<F, dynamic/*=O2*/> kill/*<O2>*/() =>
      interpret((h, t) => t.kill/*<O2>*/(),
          (req, recv) => recv(left(Kill)).drain/*<O2>*/().onHalt((e) => e == Kill ? halt(End) : halt(e)),
          halt);


  Conveyor<F, dynamic/*=O2*/> pipe/*<O2>*/(Conveyor<From<O>, dynamic/*=O2*/> c2) =>
      c2.interpret((h, t) => produce(h, pipe /*<O2>*/(t)),
          (req, recv) =>
              this.interpret((h, t) => t.pipe/*<O2>*/(Try(() => recv(right(h)))),
                  (req0, recv0) => consume/*<F, O, O2>*/(req0, (ea) => recv0(ea).pipe /*<O2>*/(c2)),
                  (err) => halt/*<F, O>*/(err).pipe/*<O2>*/(recv(left(err)))),
          (err) => kill/*<O2>*/().onHalt((err2) => halt/*<F, O2>*/(err).plus(halt(err2))));

  Conveyor<F, dynamic> operator |(Conveyor<From<O>, dynamic> c2) => pipe(c2);

  Conveyor<F, O> lastOr(O o) =>
      interpret((h, t) => t.lastOr(h),
          (req, recv) => consume(req, (ea) => recv(ea).lastOr(o)),
          (err) => err == End ? produce(o) : halt(err));

  static Conveyor/*<F, O>*/ Try/*<F, O>*/(Conveyor/*<F, O>*/ p()) {
    try {
      return p();
    } catch (err) {
      return halt(err);
    }
  }

  Conveyor/*<F, O2>*/ tryOrDie/*<O2>*/(Conveyor/*<F, O2>*/ p()) {
    try {
      return p();
    } catch (err) {
      return kill/*<O2>*/().onHalt((err2) => halt/*<F, O2>*/(err).plus(halt(err2)));
    }
  }

  Conveyor<F, O> take(int n) => pipe(Pipe.take(n));

  Conveyor<F, O> takeWhile(bool f(O o)) => pipe(Pipe.takeWhile(f));

  Conveyor<F, O> drop(int n) => pipe(Pipe.drop(n));

  Conveyor<F, O> dropWhile(bool f(O o)) => pipe(Pipe.dropWhile(f));

  Conveyor<F, O> filter(bool f(O o)) => pipe(Pipe.filter(f));

  Conveyor<F, dynamic/*=O2*/> fold/*<O2>*/(/*=O2*/ z, Function2/*<O2, O, O2>*/ f) => pipe(Pipe.scan(z, f)).lastOr(z);

  Conveyor<F, dynamic/*=O2*/> foldWhile/*<O2>*/(/*=O2*/ z, Function2/*<O2, O, O2>*/ f, Function1/*<O2, bool>*/ p) => pipe(Pipe.scanWhile(z, f, p)).lastOr(z);

  Conveyor<F, O> concatenate(Monoid<O> monoid) => pipe(Pipe.scan(monoid.zero(), monoid.append)).lastOr(monoid.zero());

{'fraction_non_alphanumeric': 0.2645430474203643, 'fraction_numerical': 0.010751825403193452, 'mean_word_length': 3.4609878310665714, 'pattern_counts': {'":': 0, '<': 231, '<?xml version=': 0, '>': 408, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1055502', 'n_tokens_mistral': 5274, 'n_tokens_neox': 5158, 'n_words': 1017}