Tweaks for MessageGrouper.

Change-Id: I1be1d7e998ee28414a455816383e5d0810d4ad9e
Reviewed-on: https://dart-review.googlesource.com/c/sdk/+/239403
Reviewed-by: Jake Macdonald <[email protected]>
Commit-Queue: Konstantin Shcheglov <[email protected]>

Author: scheglov

pkg/_fe_analyzer_shared/lib/src/macros/executor/message_grouper.dart

@@ -3,7 +3,6 @@
 // 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.
@@ -14,18 +13,11 @@ class MessageGrouper {
   /// The input bytes stream subscription.
   late final StreamSubscription _inputStreamSubscription;
 
-  /// The length of the current message to read, or `-1` if we are currently
-  /// reading the length.
-  int _length = -1;
-
   /// The buffer to store the length bytes in.
-  BytesBuilder _lengthBuffer = new BytesBuilder();
+  final _FixedBuffer _lengthBuffer = new _FixedBuffer(4);
 
   /// If reading raw data, buffer for the data.
-  Uint8List _messageBuffer = new Uint8List(0);
-
-  /// The position to write the next byte in [_messageBuffer].
-  int _messagePos = 0;
+  _FixedBuffer? _messageBuffer;
 
   late StreamController<Uint8List> _messageStreamController =
       new StreamController<Uint8List>(onCancel: () {
@@ -38,45 +30,36 @@ class MessageGrouper {
   }
 
   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);
       }
     }
@@ -89,3 +72,27 @@ class MessageGrouper {
     _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;
+  }
+}