anchor.diff:    }
   }
   
  +// TODO: move to bot?
  +// TODO: should this use pause/resume? Maybe? Likely?
  +// TODO: test!
   Stream expandStream(Stream source, Stream convert(input), {Stream onDone()}) {
     final controller = new StreamController();
   
  
positive.code:   return completer.future;
  }
  
  Stream _streamFromIterableFuture(Future<Iterable> future) {
    final controller = new StreamController();
  
  
negative.code:   return controller.stream;
  }
  
  // TODO: move to bot?
  Future _pipeStreamToController(StreamController controller, Stream input) {
    final completer = new Completer();
  
  