Fix websocket in progress calls (#7394)

* Add a max chunking size when reading bytes to prevent failure

* Add more debug information

* Update the active calls response to be better

core/websockets/resources/classes/websocket_client.php

@@ -336,14 +336,35 @@ class websocket_client {
 	// Helper function to fully read N bytes
 	private function read_bytes(int $length): ?string {
 		$data = '';
+		$max_chunk_size = stream_get_chunk_size($this->resource);
+
 		while (strlen($data) < $length) {
-			$chunk = fread($this->resource, $length - strlen($data));
-			if ($chunk === false || $chunk === '') {
-				break;
+			$remaining = $length - strlen($data);
+			$read_size = min($max_chunk_size, $remaining);
+
+			// Read maximum chunk size or what is remaining
+			$chunk = fread($this->resource, $read_size);
+
+			if ($chunk === false) {
+				echo "[ERROR] fread() failed to read stream\n";
+				return null;
+			}
+
+			if ($chunk === '') {
+				$meta = stream_get_meta_data($this->resource);
+				if (!empty($meta['timed_out'])) {
+					echo "[ERROR] Socket timed out after reading " . strlen($data) . " of $length bytes\n";
+					return null;
+				}
+				// Jitter or other read issues on the socket so wait 10 ms
+				usleep(10000);
+
+				// Try again
+				continue;
 			}
 			$data .= $chunk;
 		}
-		return strlen($data) === $length ? $data : null;
+		return $data;
 	}
 
 	public function authenticate($token_name, $token_hash) {
@@ -409,6 +430,14 @@ class websocket_client {
 	}
 }
 
+// PHP <=7.4 compatibility - Replaced in PHP 8.0+
+if (!function_exists('stream_get_chunk_size')) {
+	function stream_get_chunk_size($stream): int {
+		// For PHP versions lower then 8 we send the maximum size defined from https://php.net/stream_get_chunk_size
+		return 8192;
+	}
+}
+
 /**
  * Example usage:
  */

core/websockets/resources/classes/websocket_service.php

@@ -401,6 +401,7 @@ class websocket_service extends service {
 				$message = websocket_message::create_from_json_message($json_array);
 
 				if ($message === null) {
+					$this->warn("Message is empty");
 					return;
 				}
 
@@ -419,6 +420,7 @@ class websocket_service extends service {
 
 				// Message is from the client so check the service_name that needs to get the message
 				if (!empty($message->service_name())) {
+					$this->debug("Message is from subscriber");
 					$this->handle_client_message($subscriber, $message);
 				} else {
 					// Message does not have a service name
@@ -435,6 +437,9 @@ class websocket_service extends service {
 		foreach ($this->subscribers as $service) {
 			//when we find the service send the request
 			if ($service->service_equals($message->service_name())) {
+				//notify we found the service
+				$this->debug("Routing message to service '" . $message->service_name() . "' for topic '" . $message->topic() . "'");
+
 				//attach the current subscriber permissions so the service can verify
 				$message->permissions($subscriber->get_permissions());
 
@@ -446,6 +451,7 @@ class websocket_service extends service {
 
 				//send the modified web socket message to the service
 				$service->send((string) $message);
+
 				//continue searching for service providers
 				continue;
 			}