anchor.diff:      _messageStreamController.close();
     }
   }
  +
  +/// A buffer of fixed length.
  +class _FixedBuffer {
  +  final Uint8List bytes;
  +
  +  /// The offset in [bytes].
  +  int _offset = 0;
  +
  +  _FixedBuffer(int length) : bytes = new Uint8List(length);
  +
  +  /// Return `true` when the required number of bytes added.
  +  bool get isReady => _offset == bytes.length;
  +
  +  int operator [](int index) => bytes[index];
  +
  +  void addByte(int byte) {
  +    bytes[_offset++] = byte;
  +  }
  +
  +  /// Reset the number of added bytes to zero.
  +  void reset() {
  +    _offset = 0;
  +  }
  +}
  
positive.code: // BSD-style license that can be found in the LICENSE file.
  
  import 'dart:async';
  import 'dart:math';
  import 'dart:typed_data';
  
  /// Collects messages from an input stream of bytes.
  
negative.code:   /// The input bytes stream subscription.
    late final StreamSubscription _inputStreamSubscription;
  
    /// The buffer to store the length bytes in.
    final _FixedBuffer _lengthBuffer = new _FixedBuffer(4);
  
    /// If reading raw data, buffer for the data.
    _FixedBuffer? _messageBuffer;
  
    late StreamController<Uint8List> _messageStreamController =
        new StreamController<Uint8List>(onCancel: () {
  