anchor.diff:          _handleBytes(bytes, offset);
         }
       }
  -  }
  -
  -  /// Stop listening to the input stream for further updates, and close the
  -  /// output stream.
  -  void cancel() {
  -    _inputStreamSubscription.cancel();
  -    _messageStreamController.close();
     }
   }
   
  
positive.code: // BSD-style license that can be found in the LICENSE file.
  
  import 'dart:async';
  import 'dart:typed_data';
  
  /// Collects messages from an input stream of bytes.
  
negative.code:   /// If reading raw data, buffer for the data.
    _FixedBuffer? _messageBuffer;
  
    late final StreamController<Uint8List> _messageStreamController =
        new StreamController<Uint8List>(onCancel: () {
      _inputStreamSubscription.cancel();
    });
  
    Stream<Uint8List> get messageStream => _messageStreamController.stream;
  
    MessageGrouper(Stream<List<int>> inputStream) {
      _inputStreamSubscription = inputStream.listen(_handleBytes, onDone: cancel);
    }
  
    /// Stop listening to the input stream for further updates, and close the
    /// output stream.
    void cancel() {
      _inputStreamSubscription.cancel();
      _messageStreamController.close();
    }
  
    void _handleBytes(List<int> bytes, [int offset = 0]) {
  