feat: stream monitor updates over websocket

NiroTelemetrySnapshot.php

@@ -226,9 +226,7 @@ final class NiroTelemetrySnapshot
 
             $state = [];
             $availability = null;
-            $deadline = microtime(true) + $timeout;
-            while (microtime(true) < $deadline && ($state === [] || $availability === null)) {
-                $packet = self::readPacket($reader);
+            foreach (self::collectPackets($reader, $timeout) as $packet) {
                 if (($packet['type'] ?? 0) !== 3) {
                     continue;
                 }
@@ -398,6 +396,82 @@ final class NiroTelemetrySnapshot
         ];
     }
 
+    /** @return list<array{type: int, flags: int, payload: string}> */
+    private static function collectPackets($socket, float $timeout): array
+    {
+        stream_set_blocking($socket, false);
+        $packets = [];
+        $buffer = '';
+        $deadline = microtime(true) + $timeout;
+        $lastDataAt = null;
+
+        while (microtime(true) < $deadline) {
+            if ($lastDataAt !== null && microtime(true) - $lastDataAt >= 0.5) {
+                break;
+            }
+
+            $remaining = max(0.0, min(0.1, $deadline - microtime(true)));
+            $seconds = (int)$remaining;
+            $microseconds = (int)(($remaining - $seconds) * 1000000);
+            $read = [$socket];
+            $write = null;
+            $except = null;
+            $selected = @stream_select($read, $write, $except, $seconds, $microseconds);
+            if ($selected === false) {
+                throw new RuntimeException('mqtt_select_failed');
+            }
+            if ($selected > 0) {
+                $chunk = fread($socket, 65536);
+                if ($chunk === false) {
+                    throw new RuntimeException('mqtt_read_failed');
+                }
+                if ($chunk !== '') {
+                    $buffer .= $chunk;
+                    $lastDataAt = microtime(true);
+                    if (strlen($buffer) > self::MAX_PACKET_BYTES * 2) {
+                        throw new RuntimeException('mqtt_buffer_too_large');
+                    }
+                } elseif (feof($socket)) {
+                    break;
+                }
+            }
+
+            while (strlen($buffer) >= 2) {
+                $offset = 1;
+                $packetLength = 0;
+                $multiplier = 1;
+                do {
+                    if (!isset($buffer[$offset])) {
+                        break 2;
+                    }
+                    $digit = ord($buffer[$offset]);
+                    $packetLength += ($digit & 127) * $multiplier;
+                    $multiplier *= 128;
+                    $offset++;
+                    if ($offset > 5) {
+                        throw new RuntimeException('mqtt_remaining_length_invalid');
+                    }
+                } while (($digit & 128) !== 0);
+
+                if ($packetLength > self::MAX_PACKET_BYTES) {
+                    throw new RuntimeException('mqtt_packet_too_large');
+                }
+                if (strlen($buffer) < $offset + $packetLength) {
+                    break;
+                }
+                $header = ord($buffer[0]);
+                $packets[] = [
+                    'type' => $header >> 4,
+                    'flags' => $header & 0x0f,
+                    'payload' => substr($buffer, $offset, $packetLength),
+                ];
+                $buffer = substr($buffer, $offset + $packetLength);
+            }
+        }
+
+        return $packets;
+    }
+
     private static function readExact($socket, int $length): string
     {
         $buffer = '';