fix: prevent SSHTransport busy-loop on partial packets

lib/src/ssh_transport.dart

@@ -131,6 +131,9 @@ class SSHTransport {
   /// Guards asynchronous packet processing to preserve message order.
   var _isProcessingData = false;
 
+  /// Tracks whether new socket data was received since packet processing started.
+  var _hasNewData = false;
+
   /// Identification string sent by us without trailing \r\n. For example,
   /// "SSH-2.0-DartSSH_2.0".
   String get _localVersion => 'SSH-2.0-$version';
@@ -475,6 +478,7 @@ class SSHTransport {
   /// Callback triggered when new raw bytes are received from the socket.
   void _onSocketData(Uint8List data) {
     _buffer.add(data);
+    _hasNewData = true;
     _scheduleProcessData();
   }
 
@@ -496,6 +500,8 @@ class SSHTransport {
     }
 
     _isProcessingData = true;
+    final lengthBefore = _buffer.length;
+    _hasNewData = false;
 
     _processDataAsync().catchError((error, stackTrace) {
       if (error is SSHError) {
@@ -506,7 +512,9 @@ class SSHTransport {
     }).whenComplete(() {
       _isProcessingData = false;
       if (_buffer.isNotEmpty && !isClosed) {
-        _scheduleProcessData();
+        if (_hasNewData || _buffer.length < lengthBefore) {
+          _scheduleProcessData();
+        }
       }
     });
   }

test/src/ssh_transport_version_test.dart

@@ -43,6 +43,46 @@ void main() {
 
       client.close();
     });
+
+    test('does not busy loop on partial packet after handshake', () async {
+      final socket = _FakeSSHSocket();
+      final client = SSHClient(
+        socket,
+        username: 'demo',
+      );
+
+      // Complete the version exchange.
+      socket.addIncoming('SSH-2.0-OpenSSH_3.6.1p2\r\n');
+      await _pumpUntil(() => client.remoteVersion != null);
+
+      // Send the first 4 bytes of a packet indicating a length of 100.
+      socket.addRawIncoming(Uint8List.fromList([0, 0, 0, 100]));
+
+      // Wait a moment. If there is a microtask busy loop, the delayed future
+      // will never run and the test will timeout.
+      await Future<void>.delayed(const Duration(milliseconds: 50));
+
+      client.close();
+    });
+
+    test('reschedules processing when more data remains in the buffer',
+        () async {
+      final socket = _FakeSSHSocket();
+      final client = SSHClient(
+        socket,
+        username: 'demo',
+      );
+
+      // Send the version banner followed by some extra data in one go.
+      socket.addIncoming('SSH-2.0-OpenSSH_3.6.1p2\r\nSSH-2.0-SecondLine\r\n');
+
+      // Pump until the client processes the first version.
+      await _pumpUntil(() => client.remoteVersion != null);
+
+      expect(client.remoteVersion, 'SSH-2.0-OpenSSH_3.6.1p2');
+
+      client.close();
+    });
   });
 }
 
@@ -74,6 +114,10 @@ class _FakeSSHSocket implements SSHSocket {
     _inputController.add(Uint8List.fromList(latin1.encode(data)));
   }
 
+  void addRawIncoming(Uint8List data) {
+    _inputController.add(data);
+  }
+
   @override
   Future<void> close() async {
     if (!_doneCompleter.isCompleted) {