anchor.diff:    }
   
     void _handleBytes(List<int> bytes, [int offset = 0]) {
  -    if (_length == -1) {
  -      while (_lengthBuffer.length < 4 && offset < bytes.length) {
  +    final _FixedBuffer? messageBuffer = _messageBuffer;
  +    if (messageBuffer == null) {
  +      while (offset < bytes.length && !_lengthBuffer.isReady) {
           _lengthBuffer.addByte(bytes[offset++]);
         }
  -      if (_lengthBuffer.length >= 4) {
  -        Uint8List lengthBytes = _lengthBuffer.takeBytes();
  -        _length = lengthBytes[0] << 24 |
  -            lengthBytes[1] << 16 |
  -            lengthBytes[2] << 8 |
  -            lengthBytes[3];
  +      if (_lengthBuffer.isReady) {
  +        int length = _lengthBuffer[0] << 24 |
  +            _lengthBuffer[1] << 16 |
  +            _lengthBuffer[2] << 8 |
  +            _lengthBuffer[3];
  +        // Reset the length reading state.
  +        _lengthBuffer.reset();
  +        // Switch to the message payload reading state.
  +        _messageBuffer = new _FixedBuffer(length);
  +        _handleBytes(bytes, offset);
  +      } else {
  +        // Continue reading the length.
  +        return;
         }
  -    }
  +    } else {
  +      // Read the data from `bytes`.
  +      while (offset < bytes.length && !messageBuffer.isReady) {
  +        messageBuffer.addByte(bytes[offset++]);
  +      }
   
  -    // Just pass along `bytes` without a copy if we can, and reset our state
  -    if (offset == 0 && bytes.length == _length && bytes is Uint8List) {
  -      _length = -1;
  -      _messageStreamController.add(bytes);
  -      return;
  -    }
  -
  -    // Initialize a new buffer.
  -    if (_messagePos == 0) {
  -      _messageBuffer = new Uint8List(_length);
  -    }
  -
  -    // Read the data from `bytes`.
  -    int lenToRead = min(_length - _messagePos, bytes.length - offset);
  -    while (lenToRead-- > 0) {
  -      _messageBuffer[_messagePos++] = bytes[offset++];
  -    }
  -
  -    // If we completed a message, add it to the output stream, reset our state,
  -    // and call ourselves again if we have more data to read.
  -    if (_messagePos >= _length) {
  -      _messageStreamController.add(_messageBuffer);
  -      _length = -1;
  -      _messagePos = 0;
  -
  -      if (offset < bytes.length) {
  +      // If we completed a message, add it to the output stream.
  +      if (messageBuffer.isReady) {
  +        _messageStreamController.add(messageBuffer.bytes);
  +        // Switch to the length reading state.
  +        _messageBuffer = null;
           _handleBytes(bytes, offset);
         }
       }
  
positive.code:     _messageStreamController.close();
    }
  }
  
negative.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.
  