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:   /// If reading raw data, buffer for the data.
    _FixedBuffer? _messageBuffer;
  
    late 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);
    }
  
    void _handleBytes(List<int> bytes, [int offset = 0]) {
  
negative.code:     bytes[_offset++] = byte;
    }
  
    /// Consume at most as many bytes from [source] as required by fill [bytes].
    /// Return the number of consumed bytes.
    int addBytes(List<int> source, int offset) {
      int toConsume = math.min(source.length - offset, bytes.length - _offset);
      bytes.setRange(_offset, _offset + toConsume, source, offset);
      _offset += toConsume;
      return toConsume;
    }
  
    /// Reset the number of added bytes to zero.
    void reset() {
      _offset = 0;
  