Web终端开发：Ratchet无法向Socket客户端发送ping命令输出

我正在开发一款个人使用的terminal on web应用，采用Ratchet作为Socket服务器。ls等简单命令可以正常运行并向客户端返回结果，但执行ping这类命令时，stdout和stderr的内容无法发送到Socket客户端，不过在服务器终端中可以看到这些命令的实时输出。

原始实现代码 #

<?php

use Ratchet\MessageComponentInterface;
use Ratchet\ConnectionInterface;

// ini_set('display_errors', 1);
// ini_set('display_startup_errors', 1);
// error_reporting(E_ALL);

require 'vendor/autoload.php'; // Ensure you have Ratchet installed

class TerminalServer implements MessageComponentInterface {
    protected $clients;
    private array $open_processes;

    public function __construct() {
        $this->open_processes = [];
        $this->clients = new \SplObjectStorage;
    }

    public function onOpen(ConnectionInterface $conn) {
        echo "New connection\n";
        $this->clients->attach($conn);
    }

    static function get_output_json( string $message , array $opts = []){
        return json_encode( [ 'data' => $message , 'opts' => $opts ] );
    }

    private function command_exec(ConnectionInterface &$from , string $command){
        
        $descriptorspec = [
            0 => ["pipe", "r"],  // stdin
            1 => ["pipe", "w"],  // stdout
            2 => ["pipe", "w"],  // stderr
        ];
    
        $process = proc_open($command, $descriptorspec, $pipes);
        $this->open_processes[] = $process;
    
        if (is_resource($process)) {

            // Close the input pipe, since we're not sending any input
            fclose($pipes[0]);
    
            // Set the pipes to non-blocking mode
            stream_set_blocking($pipes[1], false);
            stream_set_blocking($pipes[2], false);
    
            while (true) {

                $output = stream_get_contents($pipes[1]);
                $error = stream_get_contents($pipes[2]);
    
                if ($output) {

                    $string_to_send = $this->get_output_json($output);
                    echo "Out: $string_to_send\n";
                    $from->send($string_to_send);

                }
                
                if ($error) {

                    $string_to_send = $this->get_output_json($error);
                    echo "Err: $string_to_send\n";
                    $from->send($string_to_send);

                }

                flush();

                // Check if the process is still running                
                $status = proc_get_status($process);
                if (!$status['running']) {
                    break;
                }
    
                // Small delay to prevent busy-waiting
                usleep(100000); // 100ms
            }
    
            // Close pipes and process
            fclose($pipes[1]);
            fclose($pipes[2]);
            $return_value = $this->command_end($process);

            $from->send($this->get_output_json("Command ended with code $return_value\n", ['exit' => true]));

        } else {

            $from->send($this->get_output_json("Command failed to execute", ['exit' => true]));

        }

    }    

    /**end or terminate a command and return the exit code */
    public static function command_end( $proc ) :int|false {

        if( !is_resource($proc) ){
            return false;
        }

        $status = proc_get_status( $proc );

        if( $status['running'] === true ) {
            proc_terminate($proc , SIGINT);
            return SIGINT;
        } else {
            return proc_close($proc);
        }

    }

    public function onMessage(ConnectionInterface $from, $msg) {
        $command = trim($msg);
        
//        $allowedCommands = ['ls -l', 'pwd', 'whoami'];
    
        if (true || in_array($command, $allowedCommands)) {
            $this->command_exec($from , $command);
        } else {            
            $from->send($this->get_output_json("Command not allowed.\r\n"));
            
        }

    }

    public function onClose(ConnectionInterface $conn) {

        echo "Connection closed\n";

        foreach ($this->open_processes as $p) {
            var_dump($p);
            proc_close($p);
        }
        $this->clients->detach($conn);

    }

    public function onError(ConnectionInterface $conn, \Exception $e) {
        echo "Connection closed because error\n";
        $conn->close();
    }
}

$app = new Ratchet\App('localhost', 2222);
$app->route('/terminal', new TerminalServer, ['*']);
echo "No errors so far\n";
$app->run();

尝试过的修改方案 #

我曾尝试修改command_exec函数，让命令读取10行后退出，此时会在命令退出后一次性发送所有输出：

private function command_exec(ConnectionInterface &$from , string $command){
        
        $descriptorspec = [
            0 => ["pipe", "r"],  // stdin
            1 => ["pipe", "w"],  // stdout
            2 => ["pipe", "w"],  // stderr
        ];
    
        $process = proc_open($command, $descriptorspec, $pipes);
        $this->open_processes[] = $process;
    
        if (is_resource($process)) {

            // Close the input pipe, since we're not sending any input
            fclose($pipes[0]);
    
            // Set the pipes to non-blocking mode
            stream_set_blocking($pipes[1], false);
            stream_set_blocking($pipes[2], false);

            $i = 0;//count line read

            while (true) {

                $output = stream_get_contents($pipes[1]);
                $error = stream_get_contents($pipes[2]);
    
                if ($output) {

                    $string_to_send = $this->get_output_json($output);
                    echo "Out: $string_to_send\n";
                    $from->send($string_to_send);
                    $i++;

                    if( $i > 10 ){
                        break;
                    }

                }
                
                if ($error) {

                    $string_to_send = $this->get_output_json($error);
                    echo "Err: $string_to_send\n";
                    $from->send($string_to_send);

                }

                flush();

                // Check if the process is still running                
                $status = proc_get_status($process);
                if (!$status['running']) {
                    break;
                }
    
                // Small delay to prevent busy-waiting
                usleep(100000); // 100ms
            }
    
            // Close pipes and process
            fclose($pipes[1]);
            fclose($pipes[2]);
            $return_value = $this->command_end($process);

            $from->send($this->get_output_json("Command ended with code $return_value\n", ['exit' => true]));

        } else {

            $from->send($this->get_output_json("Command failed to execute", ['exit' => true]));

        }

    }

内容的提问来源于stack exchange，提问作者jayshankarshahu