Transducers in Dart

transducers.dart

library transducers;

import 'dart:async';

reduce(coll, Function fn, init) {
  // Reduce function for Iterable.
  if (coll is Iterable) {
    var result = init;
    final iterator = coll.iterator;
    while (iterator.moveNext()) {
      result = fn(result, iterator.current);
      if (result is Reduced) return result.val;
    }
    return result;

  // Reduce function for Stream.
  } else if (coll is Stream) {
    var reduced;
    return coll.fold(init, (result, input) {
      if (reduced != null) return reduced;

      final res = fn(result, input);
      if (res is Reduced) {
        reduced = res.val;
        return reduced;
      } else {
        return res;
      }
    });
  }
}

/**
 * Boxing value indicating that we are done reducing.
 */
class Reduced {
  final val;
  Reduced(this.val);
}

add(obj, el) => obj..add(el);

compose(List funcs) =>
  (arg) =>
    funcs.reversed.fold(arg, (prevRes, func) => func(prevRes));


/**
 * Transducer generators.
 */
map(Function fn) =>
  (step) =>
    (result, input) =>
      step(result, fn(input));

filter(Function pred) =>
  (step) =>
    (result, input) =>
      pred(input) ? step(result, input) : result;

/**
 * Take generates a stateful transducer.
 */
take(int num) =>
  (step) {
    int k = 0;
    return (result, input) {
      if (k >= num) return new Reduced(result);
      k ++;
      return step(result, input);
    };
  };


void run(String description, Function fn) {
  print("\n");
  print(description);
  fn();
}

main() {
  final transducer = compose([
      filter((x) => x > 1),
      take(3),
      map((x) => x + 100)
  ]);

  run("iterable -> iterable", (){
    print(reduce([1,2,3,4,5], transducer(add), []));
  });

  run("stream -> stream", (){
    final sc = new StreamController();
    sc.stream.listen((_) => print("streamed $_"));
    reduce(new Stream.fromIterable([1,2,3,4,5]), transducer(add), sc.sink);
  });

  run("stream -> iterable", (){
    reduce(new Stream.fromIterable([1,2,3,4,5]), transducer(add), []).then(print);
  });

  run("iterable -> stream", (){
    final sc = new StreamController();
    sc.stream.listen((_) => print("streamed $_"));
    reduce([1,2,3,4,5], transducer(add), sc.sink);
  });
}