voryx/rxthruwayclient

RxPHP Client for WAMP v2

rxthruwayclient's Issues

Rapid RPC calls may hang backlogged RPCs

<?php

use function EventLoop\addPeriodicTimer;
use Rx\Observable;
use Rx\Thruway\Client;
use Thruway\Message\ResultMessage;

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

$client = new Client('ws://192.168.101.5:9090/', 'bench');

// if you add this subscription to hold things open - everything works fine
//$client->topic('some.topic')->subscribe();

$call = $client->call('some.rpc', [2, 2]);

$calls = 0;

addPeriodicTimer(1, function () use (&$calls) {
    echo "--" . $calls . "--\n";
});

Observable::interval(10)
    ->take(100)
    ->concat(Observable::of(1))
    ->flatMap(function () use ($call, &$calls) {
        $calls++;
        return $call
            ->map(function (ResultMessage $result) use (&$calls) {
                $calls--;
                return $result->getArguments()[0];
            });
    })
    ->subscribe(function ($x) use ($calls) {
        static $count = 0;

        $count++;
        if ($count % 10 == 0) {
            echo $count . " - " . $calls . "\n";
        }
    });

Exceptions get swallowed

You should be able to control exceptions the same was as the non-Rx client by throwing WampErrorException

`publish` should warn if there is an error

publish currently is quiet when an error is emitted from the observable stream. This should probably warn the user that an error happened and remind them that they need to catch all errors if they don't want to see the warning.

Q: How to debug registration handlers?

I have trouble understanding this:

If the Registration Handler throws an exception, thruway.error.invocation_exception is returned to the caller. If you would like to allow more specific error messages, you must throw a WampErrorException or, if using observable sequences that are returned from the RPC, you can onError a WampErrorException.

Where and how do I get to see if a RPC is causing an invocation_exception?
I tried

try {
    client->register('test', function($test) {
        failplz(); // Function does not exist anywhere so this should cause problems
        return;
    }
} catch (Thruway\WampErrorExceptio $e) {
    error_log_(print_r($e,true));
}

to no avail. Neither does anything happen when starting the client, nor does anything happen when I call the RPC from somewhere, I always end up with the barebone invocation_exception. An example would be very much appreciated, I'm fairly certain I'm just missing something obvious. First time dealing with ReactPHP (even more so Rx) either, so that makes me easy to confuse right now.

Throwable is not caught in RegisterObservable

RxThruwayClient/src/Observable/RegisterObservable.php

Line 105 in 85a8b52

} catch (\Exception $e) {

Is there a way to make a signle call and return value?

Need a composer.lock file or a fix to work with current version of dependencies

Hello, I just want to report that since there is no composer.lock file in this package, trying to use the RxThruwayClient with the current version of dependencies will raise an error:

PHP Fatal error:  Uncaught Exception: Please set a default scheduler factory in /mnt/c/DEV/RxThruwayClient/vendor/reactivex/rxphp/src/Scheduler.php:24
Stack trace:
#0 /mnt/c/DEV/RxThruwayClient/vendor/reactivex/rxphp/src/Subject/ReplaySubject.php(52): Rx\Scheduler::getDefault()
#1 /mnt/c/DEV/RxThruwayClient/src/Subject/WebSocketSubject.php(31): Rx\Subject\ReplaySubject->__construct()
#2 /mnt/c/DEV/RxThruwayClient/src/Client.php(40): Rx\Thruway\Subject\WebSocketSubject->__construct()
#3 /mnt/c/DEV/RxThruwayClient/examples/subscribe.php(8): Rx\Thruway\Client->__construct()
#4 {main}
  thrown in /mnt/c/DEV/RxThruwayClient/vendor/reactivex/rxphp/src/Scheduler.php on line 24

Today I updated my dependencies and noticed this problem, for now I am downgrading the dependencies to their previous version.

Loading composer repositories with package information
Updating dependencies
Lock file operations: 0 installs, 8 updates, 1 removal
  - Removing voryx/event-loop (3.0.2)
  - Upgrading react/dns (v1.11.0 => v1.12.0)
  - Upgrading react/promise (v2.10.0 => v2.11.0)
  - Upgrading react/promise-stream (v1.6.0 => v1.7.0)
  - Upgrading react/socket (v1.14.0 => v1.15.0)
  - Upgrading rx/websocket (2.1.8 => 2.2.0)
  - Upgrading seba1rx/sessionadmin (dev-main 8bdecda => dev-main 4e86497)
  - Upgrading symfony/http-foundation (v6.3.8 => v6.4.2)
  - Upgrading symfony/routing (v6.3.5 => v6.4.2)
Writing lock file
Installing dependencies from lock file (including require-dev)
Package operations: 0 installs, 8 updates, 1 removal
  - Downloading symfony/routing (v6.4.2)
  - Downloading symfony/http-foundation (v6.4.2)
  - Downloading react/promise-stream (v1.7.0)
  - Downloading rx/websocket (2.2.0)
  - Downloading seba1rx/sessionadmin (dev-main 4e86497)
  - Removing voryx/event-loop (3.0.2)
  - Upgrading symfony/routing (v6.3.5 => v6.4.2): Extracting archive
  - Upgrading symfony/http-foundation (v6.3.8 => v6.4.2): Extracting archive
  - Upgrading react/promise (v2.10.0 => v2.11.0): Extracting archive
  - Upgrading react/dns (v1.11.0 => v1.12.0): Extracting archive
  - Upgrading react/socket (v1.14.0 => v1.15.0): Extracting archive
  - Upgrading react/promise-stream (v1.6.0 => v1.7.0): Extracting archive
  - Upgrading rx/websocket (2.1.8 => 2.2.0): Extracting archive
  - Upgrading seba1rx/sessionadmin (dev-main 8bdecda => dev-main 4e86497): Extracting archive

I used the following to downgrade the affected dependecies in order to make it work again:

 composer require voryx/event-loop 3.0.2
 composer require rect/dns 1.11.0
 composer require react/dns 1.11.0
 composer require react/promise 2.10.0
 composer require react/promise-stream 1.6.0
 composer require react/socket 1.14.0
 composer require rx/websocket 2.1.8
 composer require symfony/http-foundation 6.3.8
 composer require symfony/routing 6.3.5

Is it possible for you to provide a fix for this or a composer.lock file ?

Question/Help - Waiting for call completion?

I have a custom process that runs both RxThruwayClient and bunnyPHP's async client (reactPHP based), using a single eventloop. This process is the business logic serverside and works as a job manager (among other things).

The users will call the job scheduler (rxthruway) and expect a simple result: the id value the generated job will have. But before this can be returned, the rxthruwayclient has to finish making its own RPC call to the internal AuthorizationManager (add_authorization_rule), as the client will then subscribe to that specific topic to receive information about the job status.
How can I implement this?

This is my first project dealing with ReactPHP and Rx/Observable, so I fear I might make very basic mistakes.

Simplified code sample:

        $client->register('example.slowjob', function ($jobdata) {
                $jobid = uniqid();
                $client->call('add_authorization_rule', [(object)[
                    'role'  => 'user',
                    'action'=> 'subscribe',
                    'uri'   => 'example.jobstatus.' . $jobid,
                    'allow' => true
                ]])->subscribe(
                    function (ResultMessage $res) {
                       // This is where the RPC return $jobid should be triggered.
                });
                return $jobid; // this should only be returned after above call is complete. this way, it gets returned too early, leading to the user trying tosubscribe before he is allowed to.
        })->subscribe();

The only solution I could come up with was to publish on a different, already-subscribed topic when the add_auth-call is complete, to which the client can then react by subscribing to the new topic. But that seems unnecessarily convoluted, and would make the RPC pointless (as it doesn't return anything of value).
I obviously want to avoid blocking the event loop, but I feel there must be an easier way.

Help would be appreciated!

Hostname resolution for thruway client

When creating a new Rx\Thruway\Client, I cannot use hostnames, but must use IP address.

Why is that?

This works:
ws://123.456.78.9:9090

this doesn't:
ws://myservername:9090

My current workaround is to use
'ws://' . gethostbyname('myservername') . ':9090',

which feels wrong and unreliable.
If it's the same machine, using localhost:9090 also works, just not the actual hostname.

I use the same URL (incl. hostname) in vxWamp (used on browser/clientside) to reach the server, which works fine, which makes me believe it's not a resolution issue.

Type should be Throwable here

RxThruwayClient/src/Client.php

Line 207 in 71dd23e

->flatMap(function (\Exception $ex) use ($maxRetryDelay, $retryDelayGrowth, $initialRetryDelay) {

Update Pawl to v0.3.0

This package is no longer compatible with main thruway because both require different ratchet/pawl versions, making it impossible to run the thruway router and this client at the same time.

Result of composer require rx/thruway-client

Using version ^3.0 for rx/thruway-Client
./composer.json has been updated
Loading composer repositories with package information
Updating dependencies (including require-dev)
Your requirements could not be resolved to an installable set of packages.

Problem 1
- Conclusion: don't install rx/thruway-client 3.0.2
- Conclusion: don't install rx/thruway-client 3.0.1
- Conclusion: remove ratchet/pawl v0.3.1
- Installation request for rx/thruway-client ^3.0 -> satisfiable by rx/thruway-client[3.0.0, 3.0.1, 3.0.2].
- Conclusion: don't install ratchet/pawl v0.3.1
- rx/thruway-client 3.0.0 requires ratchet/pawl ^0.2.2 -> satisfiable by ratchet/pawl[v0.2.2, v0.2.3].
- Can only install one of: ratchet/pawl[v0.2.2, v0.3.1].
- Can only install one of: ratchet/pawl[v0.2.3, v0.3.1].
- Installation request for ratchet/pawl (locked at v0.3.1) -> satisfiable by ratchet/pawl[v0.3.1].

Installation failed, reverting ./composer.json to its original content.

Improve error reported when callable signature is wrong

When the callable on a registration does not have the correct signature, the only error you get back is thruway.error.invocation_exception. This should be improved.