anchor.diff:      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;
  
positive.code:         _handleBytes(bytes, offset);
        }
      }
    }
  
    /// Stop listening to the input stream for further updates, and close the
    /// output stream.
    void cancel() {
      _inputStreamSubscription.cancel();
      _messageStreamController.close();
    }
  }
  
  
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]) {
  