IteratorStream.hx

class IteratorStream<Item, Quality> extends StreamBase<Item, Quality> {
	var iterator: Iterator<Item>;

	public function new(iterator)
		this.iterator = iterator;

	override function next():Future<Step<Item, Quality>>
		return switch iterator.hasNext() {
			case true: Link(iterator.next(), this);
			case false: End;
		}

	override public function forEach<Safety>(handler:Handler<Item, Safety>) {
		return Future.async(function step(cb:Conclusion<Item, Safety, Quality>->Void) 
			switch iterator.hasNext() {
				case true: 
					handler.apply(iterator.next()).handle(function (s) {
						switch s {
							case BackOff:
								cb(Halted(this));
							case Finish:
								cb(Halted(this));
							case Resume:
								step(cb);
							case Clog(e):
								cb(Clogged(e, this));
						}
					});
				case false: cb(Depleted);
			}, true);
	}
}