anchor.diff:      Completer streamDone = new Completer();
       _sub = stream.listen((items) {
         _chunks.add(items);
  -      _counter += items is List ? (items as List).length : 1;
  +      _counter += items is List ? items.length : 1;
         if (limited && _counter >= limit) {
           _sub.pause();
         }
  
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:   }
  
    @override
    Future addStream(Stream<dynamic/*T|List<T>*/> stream) {
      var lastStream = _currentStream == null ? stream : _currentStream;
      if (_sub != null) {
        _sub.cancel();
  