amphp/parallel + ssh2

Inizia con Main.php e SSHExecTask.php, quindi traccia il modo in cui l’oggetto SSH viene passato a ContextWorkerPool e come viene utilizzata la sua connessione in src/SSH.php. Riproduci il malfunzionamento con l’esempio fornito e ispeziona il confine del worker e il valore della risorsa segnalato. L’attività è completata quando il task viene eseguito senza il TypeError e restituisce la risposta del comando.

Hello everyone! I like and very appreciate your work! Great library and docs!

I am trying to use ssh2 extension with parallel. Unfortunately, I am getting a mysterious issue. When I am calling a method in the object of my SSH class (for example, exec), the connection variable, initialized in the constructor, always resets to a "0" value instead of a resource. Full error looks like that:

Uncaught TypeError thrown in context with message "ssh2_exec(): Argument #1 ($session) must be of type resource, int given" and code "0" in SSH.php:18

What I am doing wrong? Thank you!

Main.php:

<?php
error_reporting( E_ALL & ~E_DEPRECATED );

use Amp\Parallel\Worker\ContextWorkerPool;
use Par\SSHExecTask;

require_once __DIR__ . '/vendor/autoload.php';

$pool = new ContextWorkerPool( 1 );
$ssh = new \Par\SSH( '127.0.0.1', 322, 'user', '123456' );
$task = $pool->submit( new SSHExecTask( $ssh ) );

$response = $task->await();
var_dump( $response );

src/SSH.php:

<?php
namespace Par;

class SSH
{
    private $connection;
    private $sftp;

    public function __construct( $host, $port, $user, $password )
    {
        $this->connection = ssh2_connect( $host, $port );
        ssh2_auth_password( $this->connection, $user, $password );
        $this->sftp = ssh2_sftp( $this->connection );
    }

    public function exec( $command )
    {
        $stdout = ssh2_exec( $this->connection, $command );
        $stderr = ssh2_fetch_stream( $stdout, SSH2_STREAM_STDERR );

        stream_set_blocking( $stdout, true );
        stream_set_blocking( $stderr, true );

        $stdoutContent = stream_get_contents( $stdout );
        $stderrContent = stream_get_contents( $stderr );

        if( $stderrContent ) {
            throw new \RuntimeException( $stderrContent );
        }

        fclose( $stdout );
        fclose( $stderr );

        return trim( $stdoutContent );
    }

    public function upload( $localFile, $remoteFile )
    {
        $stream = fopen( "ssh2.sftp://{$this->sftp}/$remoteFile", 'w' );
        return fwrite( $stream, file_get_contents( $localFile ) );
    }

    public function mkdir( $dirname, $mod = 0777, $recursive = false )
    {
        return ssh2_sftp_mkdir( $this->sftp, $dirname, $mod, $recursive );
    }
}

src/SSHExecTask.php:

<?php
namespace Par;

use Amp\Cancellation;
use Amp\Parallel\Worker\Task;
use Amp\Sync\Channel;

class SSHExecTask implements Task
{
    public function __construct(
        private readonly SSH $ssh
    ) {}

    public function run( Channel $channel, Cancellation $cancellation ): bool
    {
        return $this->ssh->exec( 'echo 123' );
    }
}

composer.json:

{
    "require": {
        "amphp/parallel": "^2.1",
        "ext-ssh2": "*"
    },
    "autoload": {
        "psr-4": {
            "Par\\": "src"
        }
    }
}