anchor.diff:    StreamSubscription listen(void onData(value),
         {void onError(var error), void onDone(),
          bool cancelOnError}) {
  -    return _controller.stream.listen(onData,
  +    return _stream.listen(onData,
           onError: onError,
           onDone: onDone,
           cancelOnError: true);
  
positive.code:   /// The underlying [StreamController] for [this].
    final StreamController _controller;
  
    /// The [StreamSubscription] that connects the wrapped [Stream] to
    /// [_controller].
    StreamSubscription _subscription;
  
negative.code:   /// Creates a new [_ErrorGroupFuture] that's a child of [_group] and wraps
    /// [inner].
    _ErrorGroupStream(this._group, Stream inner)
      : _controller = new StreamController(sync: true) {
      // Use old-style asBroadcastStream behavior - cancel source _subscription
      // the first time the stream has no listeners.
      _stream = inner.isBroadcast
          ? _controller.stream.asBroadcastStream(onCancel: (sub) => sub.cancel())
          : _controller.stream;
      _subscription = inner.listen((v) {
        _controller.add(v);
      }, onError: (e) {
  