anchor.diff:    }
   
     @override
  -  Future addStream(Stream<T> stream) {
  +  Future addStream(Stream<dynamic/*T|List<T>*/> stream) {
       var lastStream = _currentStream == null ? stream : _currentStream;
       if (_sub != null) {
         _sub.cancel();
  
positive.code:  * Throws [UnderflowError] if [throwOnError] is true. Useful for unexpected
   * [Socket] disconnects.
   */
  class StreamBuffer<T> implements StreamConsumer<T> {
    List _chunks = [];
    int _offset = 0;
    int _counter = 0; // sum(_chunks[*].length) - _offset
    List<_ReaderInWaiting<List<T>>> _readers = [];
    StreamSubscription<T> _sub;
    Completer _streamDone;
  
    final bool _throwOnError;
  
negative.code:     Completer streamDone = new Completer();
      _sub = stream.listen((items) {
        _chunks.add(items);
        _counter += items is List ? items.length : 1;
        if (limited && _counter >= limit) {
          _sub.pause();
        }
  