ApiClients\Client\Pusher\WebSocket

Complexity

Total Complexity

6

Size/Duplication

Total Lines	58
Duplicated Lines	0 %

Coupling/Cohesion

Components	1
Dependencies	6

Test Coverage

Coverage

100%

Importance

Changes

0

4 Methods

Rating	Name	Duplication	Size	Complexity
A	__construct()	0	5	1
A	createFactory()	0	17	3
A	onNext()	0	4	1
A	_subscribe()	0	18	1

<?php declare(strict_types=1);

namespace ApiClients\Client\Pusher;

use PackageVersions\Versions;
use React\Dns\Resolver\Resolver;
use React\EventLoop\LoopInterface;
use React\Socket\Connector;
use Rx\DisposableInterface;
use Rx\Observable;
use Rx\ObserverInterface;
use Rx\Subject\ReplaySubject;
use Rx\Subject\Subject;
use Rx\Websocket\Client;

This use statement conflicts with another class in this namespace, ApiClients\Client\Pusher\Client.

/**
 * Class WebSocket - WebSocket wrapper that queues messages while the connection is being established.
 */
final class WebSocket extends Subject
{
    private $ws;
    private $sendSubject;

    /**
     * @internal
     * @param  Observable                $ws
     * @throws \InvalidArgumentException
     */
    public function __construct(Observable $ws)
    {
        $this->sendSubject = new ReplaySubject();
        $this->ws = $ws;
    }

    public static function createFactory(string $url, bool $useMessageObject = false, array $subProtocols = [], LoopInterface $loop = null, Resolver $resolver = null): Subject
    {
        if (Versions::getVersion('rx/websocket')[0] === '1' || $resolver === null) {
            return new self(
                new Client($url, $useMessageObject, $subProtocols, $loop, $resolver)

It seems like $resolver defined by parameter $resolver on line 35 can also be of type object<React\Dns\Resolver\Resolver>; however, Rx\Websocket\Client::__construct() does only seem to accept null|object<React\Socket\ConnectorInterface>, maybe add an additional type check?

            );
        }

        return new self(
            new Client($url, $useMessageObject, $subProtocols, $loop, new Connector(
                $loop,

It seems like $loop defined by parameter $loop on line 35 can be null; however, React\Socket\Connector::__construct() does not accept null, maybe add an additional type check?

                [
                    'dns' => $resolver,
                ]
            ))
        );
    }

    public function onNext($value): void
    {
        $this->sendSubject->onNext($value);
    }

    protected function _subscribe(ObserverInterface $observer): DisposableInterface
    {
        return $this->ws
            ->do(function (Subject $ms): void {
                // Replay buffered messages onto the MessageSubject
                $this->sendSubject->subscribe($ms);

                // Now that the connection has been established, use the message subject directly.
                $this->sendSubject = $ms;
            })
            ->finally(function (): void {
                // The connection has closed, so start buffering messages util it reconnects.
                $this->sendSubject = new ReplaySubject();
            })
            ->switch()
            ->repeatDelay()
            ->subscribe($observer);
    }
}