Receive chunking

Author: abnegate

src/Client.php

@@ -377,9 +377,9 @@ public function send(mixed $data): stdClass|array|int
      */
     private function receive(): stdClass|array|int
     {
+        $chunks = [];
         $receivedLength = 0;
         $responseLength = null;
-        $res = '';
         $attempts = 0;
         $maxAttempts = 10000;
         $sleepTime = 100;
@@ -407,26 +407,37 @@ private function receive(): stdClass|array|int
             $attempts = 0;
             $sleepTime = 100; // Reset to 0.1ms
 
-            $receivedLength += \strlen($chunk);
-            $res .= $chunk;
+            $chunkLen = \strlen($chunk);
+            $receivedLength += $chunkLen;
+            $chunks[] = $chunk;
 
             // Parse message length from first 4 bytes
-            if ((!isset($responseLength)) && (strlen($res) >= 4)) {
-                $responseLength = \unpack('Vl', substr($res, 0, 4))['l'];
+            if ($responseLength === null && $receivedLength >= 4) {
+                $firstData = $chunks[0];
+
+                if (\strlen($firstData) < 4) {
+                    $firstData = \implode('', $chunks);
+                }
+
+                $responseLength = \unpack('Vl', substr($firstData, 0, 4))['l'];
 
                 // Validate response length before allocating memory to prevent memory exhaustion
                 if ($responseLength > 16777216) { // 16MB limit
                     throw new Exception('Response too large: ' . $responseLength . ' bytes');
                 }
 
-                // Additional validation for negative or tiny values
+                // Validate for negative or tiny values
                 if ($responseLength < 21) { // Minimum MongoDB message size
                     throw new Exception('Invalid response length: ' . $responseLength . ' bytes');
                 }
             }
-        } while (
-            (!isset($responseLength)) || ($receivedLength < $responseLength)
-        );
+
+            if ($responseLength !== null && $receivedLength >= $responseLength) {
+                break;
+            }
+        } while (true);
+
+        $res = \implode('', $chunks);
 
         return $this->parseResponse($res, $responseLength);
     }
@@ -1569,25 +1580,25 @@ private function parseResponse(string $response, int $responseLength): stdClass|
          * These 21 bytes are protocol metadata and precede the actual BSON-encoded document in the response.
          */
 
-        if (strlen($response) < 21) {
+        if (\strlen($response) < 21) {
             throw new Exception('Invalid response: too short');
         }
 
         // Extract message header
-        $header = substr($response, 0, 16);
-        $headerData = unpack('VmessageLength/VrequestID/VresponseTo/VopCode', $header);
+        $header = \substr($response, 0, 16);
+        $headerData = \unpack('VmessageLength/VrequestID/VresponseTo/VopCode', $header);
 
         // Validate message length
         if ($headerData['messageLength'] !== $responseLength) {
             throw new Exception('Response length mismatch');
         }
 
         // Extract flag bits and payload type
-        $flagBits = unpack('V', substr($response, 16, 4))[1];
-        $payloadType = ord(substr($response, 20, 1));
+        $flagBits = \unpack('V', \substr($response, 16, 4))[1];
+        $payloadType = \ord(\substr($response, 20, 1));
 
         // Extract BSON document (skip header + flagBits + payloadType)
-        $bsonString = substr($response, 21, $responseLength - 21);
+        $bsonString = \substr($response, 21, $responseLength - 21);
 
         if (empty($bsonString)) {
             return new \stdClass();
@@ -1598,32 +1609,32 @@ private function parseResponse(string $response, int $responseLength): stdClass|
             $result = Document::fromBSON($bsonString)->toPHP();
 
             // Convert array to stdClass if needed
-            if (is_array($result)) {
+            if (\is_array($result)) {
                 $result = (object)$result;
             }
 
             // Check for write errors (duplicate key, etc.)
-            if (property_exists($result, 'writeErrors') && !empty($result->writeErrors)) {
+            if (\property_exists($result, 'writeErrors') && !empty($result->writeErrors)) {
                 throw new Exception(
                     $result->writeErrors[0]->errmsg,
                     $result->writeErrors[0]->code
                 );
             }
 
             // Check for general MongoDB errors
-            if (property_exists($result, 'errmsg')) {
+            if (\property_exists($result, 'errmsg')) {
                 throw new Exception(
                     'E' . $result->code . ' ' . $result->codeName . ': ' . $result->errmsg,
                     $result->code
                 );
             }
 
             // Check for operation success
-            if (property_exists($result, 'n') && $result->ok === 1.0) {
+            if (\property_exists($result, 'n') && $result->ok === 1.0) {
                 return $result->n;
             }
 
-            if (property_exists($result, 'nonce') && $result->ok === 1.0) {
+            if (\property_exists($result, 'nonce') && $result->ok === 1.0) {
                 return $result;
             }
 
@@ -1636,7 +1647,7 @@ private function parseResponse(string $response, int $responseLength): stdClass|
             throw new Exception('Failed to parse BSON response: ' . $e->getMessage());
         } catch (\Exception $e) {
             if ($e instanceof Exception) {
-                throw $e; // Re-throw our own exceptions
+                throw $e;
             }
             throw new Exception('Error parsing response: ' . $e->getMessage());
         }
@@ -1671,11 +1682,11 @@ public function isTransientTransactionError(Exception $exception): bool
             13436, // NotMasterOrSecondary
         ];
 
-        return in_array($code, $transientCodes) ||
-            str_contains($message, self::TRANSIENT_TRANSACTION_ERROR) ||
-            str_contains($message, 'connection') ||
-            str_contains($message, 'timeout') ||
-            str_contains($message, 'network');
+        return \in_array($code, $transientCodes) ||
+            \str_contains($message, self::TRANSIENT_TRANSACTION_ERROR) ||
+            \str_contains($message, 'connection') ||
+            \str_contains($message, 'timeout') ||
+            \str_contains($message, 'network');
     }
 
     /**
@@ -1702,8 +1713,8 @@ public function isUnknownTransactionCommitResult(Exception $exception): bool
             13436, // NotMasterOrSecondary
         ];
 
-        return in_array($code, $unknownCommitCodes) ||
-            str_contains($message, self::UNKNOWN_TRANSACTION_COMMIT_RESULT);
+        return \in_array($code, $unknownCommitCodes) ||
+            \str_contains($message, self::UNKNOWN_TRANSACTION_COMMIT_RESULT);
     }
 
     /**
@@ -1769,7 +1780,7 @@ public function withTransaction(array $session, callable $callback, array $optio
                 if ($this->isTransientTransactionError($e) && $attempt < $maxRetries) {
                     $attempt++;
                     if ($retryDelayMs > 0) {
-                        usleep($retryDelayMs * 1000);
+                        \usleep($retryDelayMs * 1000);
                     }
                     continue;
                 }