anchor.diff:        return request.future;
       }
   
  -    if (_isDone) return new Future.value();
  -    if (_subscription == null) _subscription = _sourceStream.listen(null);
  -    var future = _subscription.cancel();
  -    _onDone();
  -    return future;
  -  }
  +    if (_isDone && _eventQueue.isEmpty) return new Future.value();
  +    return _cancel();
  +  }
  +
  +  // ------------------------------------------------------------------
  +  // Methods that may be called from the request implementations to
  +  // control the even stream.
  +
  +  /// Matches events with requests.
  +  ///
  +  /// Called after receiving an event or when the event source closes.
  +  ///
  +  /// May be called by requests which have returned `false` (saying they
  +  /// are not yet done) so they can be checked again before any new
  +  /// events arrive.
  +  /// Any request returing `false` from `update` when `isDone` is `true`
  +  /// *must* call `_updateRequests` when they are ready to continue
  +  /// (since no further events will trigger the call).
  +  void _updateRequests() {
  +    while (_requestQueue.isNotEmpty) {
  +      if (_requestQueue.first.update(_eventQueue, _isDone)) {
  +        _requestQueue.removeFirst();
  +      } else {
  +        return;
  +      }
  +    }
  +
  +    if (!_isDone) {
  +      _pause();
  +    }
  +  }
  +
  +  /// Extracts a stream from the event source and makes this stream queue
  +  /// unusable.
  +  ///
  +  /// Can only be used by the very last request (the stream queue must
  +  /// be closed by that request).
  +  /// Only used by [rest].
  +  Stream _extractStream();
  +
  +  /// Requests that the event source pauses events.
  +  ///
  +  /// This is called automatically when the request queue is empty.
  +  ///
  +  /// The event source is restarted by the next call to [_ensureListening].
  +  void _pause();
  +
  +  /// Ensures that we are listening on events from the event source.
  +  ///
  +  /// Starts listening for the first time or resumes after a [_pause].
  +  ///
  +  /// Is called automatically if a request requires more events.
  +  void _ensureListening();
  +
  +  /// Cancels the underlying event source.
  +  Future _cancel();
  +
  +  // ------------------------------------------------------------------
  +  // Methods called by the event source to add events or say that it's
  +  // done.
  +
  +  /// Called when the event source adds a new data or error event.
  +  /// Always calls [_updateRequests] after adding.
  +  void _addResult(Result result) {
  +    _eventQueue.add(result);
  +    _updateRequests();
  +  }
  +
  +  /// Called when the event source is done.
  +  /// Always calls [_updateRequests] after adding.
  +  void _close() {
  +    _isDone = true;
  +    _updateRequests();
  +  }
  +
  +  // ------------------------------------------------------------------
  +  // Internal helper methods.
   
     /// Returns an error for when a request is made after cancel.
     ///
  
positive.code:     _completer.complete(_list);
      return true;
    }
  
    void close(Queue<Result> events) {
      _completer.complete(_list);
    }
  }
  
  /// Request for a [StreamQueue.cancel] call.
  
negative.code:   /// This method is called when a request reaches the front of the request
    /// queue, and if it returns `false`, it's called again every time a new event
    /// becomes available, or when the stream closes.
    /// If the function returns `false` when the stream has already closed
    /// ([isDone] is true), then the request must call
    /// [StreamQueue._updateRequests] itself when it's ready to continue.
    bool update(Queue<Result> events, bool isDone);
  }
  
  /// Request for a [StreamQueue.next] call.
  