AsyncClient::send() - Code Metrics - Inspection of "Refactored AsyncClient so that it's more declarati..." - php-api-clients/pusher - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#7)

by David

created 2017-09-05 22:12 UTC

AsyncClient::send() A

↳ Parent: AsyncClient

Complexity

Conditions	1
Paths	1

Size

Total Lines	4
Code Lines	2

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	3
CRAP Score	1

Importance

Changes

Metric	Value
dl	0
loc	4
ccs	3
cts	3
cp	1
rs	10
c	0
b	0
f	0
cc	1
eloc	2
nc	1
nop	1
crap	1

<?php declare(strict_types=1);

namespace ApiClients\Client\Pusher;

use React\Dns\Resolver\Resolver;
use React\EventLoop\LoopInterface;
use RuntimeException;
use Rx\Observable;
use Rx\Scheduler;
use Rx\Subject\Subject;
use Rx\Websocket\WebsocketErrorException;
use Throwable;

final class AsyncClient
{
    const DEFAULT_DELAY = 200;
    const NO_ACTIVITY_TIMEOUT = 120;
    const NO_PING_RESPONSE_TIMEOUT = 30;

    /**
     * @var Observable
     */
    private $messages;

    /**
     * @var array
     */
    private $channels = [];

    /**
     * @var int
     */
    private $delay = self::DEFAULT_DELAY;

    /**
     * @var Subject
     */
    private $client;

    /**
     * @var Observable
     */
    private $connected;

    /**
     * @internal
     * @param  Subject                   $client
     * @throws \InvalidArgumentException
     */
    public function __construct(Subject $client)
    {
        $this->client = $client;

        /** @var Observable $events */
        $events = $client
            ->_ApiClients_jsonDecode()
            ->map([Event::class, 'createFromMessage'])
            ->singleInstance();

        $pusherErrors = $events
            ->filter([Event::class, 'isError'])
            ->flatMap(function (Event $event) {
                $throwable = new PusherErrorException($event->getData()['message'], (int)$event->getData()['code']);

                return Observable::error($throwable);
            });

        $this->connected = $events
            ->filter([Event::class, 'connectionEstablished'])
            ->singleInstance();

        $this->messages = $events
            ->merge($this->timeout($events))
            ->merge($pusherErrors)
            ->retryWhen(function (Observable $errors) {
                return $errors->flatMap(function (Throwable $throwable) {
                    return $this->handleLowLevelError($throwable);
                });
            })
            ->singleInstance();
    }

    /**
     * @param  LoopInterface             $loop
     * @param  string                    $app      Application ID
     * @param  Resolver                  $resolver Optional DNS resolver
     * @throws \InvalidArgumentException
     * @return AsyncClient
     */
    public static function create(LoopInterface $loop, string $app, Resolver $resolver = null): AsyncClient
    {
        try {
            Scheduler::setDefaultFactory(function () use ($loop) {
                return new Scheduler\EventLoopScheduler($loop);
function acceptsInteger($int) { }

$x = '123'; // string "123"

// Instead of
acceptsInteger($x);

// we recommend to use
acceptsInteger((integer) $x);
            });
        } catch (Throwable $t) {

        }

        return new self(
            new Websocket(ApiSettings::createUrl($app), false, [], $loop, $resolver)
        );
    }

    /**
     * Listen on a channel.
     *
     * @param  string                    $channel Channel to listen on
     * @throws \InvalidArgumentException
     * @return Observable
     */
    public function channel(string $channel): Observable
    {
        // Only join a channel once
        if (isset($this->channels[$channel])) {
            return $this->channels[$channel];
        }

        // Ensure we only get messages for the given channel
        $channelMessages = $this->messages->filter(function (Event $event) use ($channel) {
            return $event->getChannel() !== '' && $event->getChannel() === $channel;
        });

        $subscribe = $this->connected
            ->do(function () use ($channel) {
                // Subscribe to pusher channel after connected
                $this->send(Event::subscribeOn($channel));
            })
            ->flatMapTo(Observable::empty());

        // Observable representing channel events
        $events = $channelMessages
            ->merge($subscribe)
            ->filter([Event::class, 'subscriptionSucceeded'])
            ->finally(function () use ($channel) {
                // Send unsubscribe event
                $this->send(Event::unsubscribeOn($channel));

                // Remove our channel from the channel list so we don't resubscribe in case we reconnect
                unset($this->channels[$channel]);
            });

        $this->channels[$channel] = $events;

        return $this->channels[$channel];
    }

    /**
     * Send a message through the client.
     *
     * @param array $message Message to send, will be json encoded
     *
     */
    public function send(array $message)
    {
        $this->client->onNext(json_encode($message));
    }

    /**
     * Returns an observable of TimeoutException.
     * The timeout observable will get cancelled every time a new event is received.
     *
     * @param  Observable $events
     * @return Observable
     */
    public function timeout(Observable $events): Observable
    {
        $timeoutDuration = $this->connected->map(function (Event $event) {
            return ($event->getData()['activity_timeout'] ?? self::NO_ACTIVITY_TIMEOUT) * 1000;
        });

        return $timeoutDuration
            ->combineLatest([$events])
            ->pluck(0)
            ->concat(Observable::of(-1))
            ->flatMapLatest(function (int $time) {

                // If the events observable ends, return an empty observable so we don't keep the stream alive
                if ($time === -1) {
                    return Observable::empty();
                }

                return Observable::never()
                    ->timeout($time)
                    ->catch(function () use ($time) {
                        // ping (do something that causes incoming stream to get a message)
                        $this->send(Event::ping());
                        // this timeout will actually timeout with a TimeoutException - causing
                        // everything above this to dispose
                        return Observable::never()->timeout($time);
                    });
            });
    }

    /**
     * Handle errors as described at https://pusher.com/docs/pusher_protocol#error-codes.
     * @param  Throwable  $throwable
     * @return Observable
     */
    private function handleLowLevelError(Throwable $throwable): Observable
    {
        // Only allow certain, relevant, exceptions
        if (!($throwable instanceof WebsocketErrorException) &&
            !($throwable instanceof RuntimeException) &&
            !($throwable instanceof PusherErrorException)
        ) {
            return Observable::error($throwable);
        }

        $code = $throwable->getCode();
        $pusherError = ($throwable instanceof WebsocketErrorException || $throwable instanceof PusherErrorException);

        // Errors 4000-4099, don't retry connecting
        if ($pusherError && $code >= 4000 && $code <= 4099) {
            return Observable::error($throwable);
        }

        // Errors 4100-4199 reconnect after 1 or more seconds, we do it after 1.001 second
        if ($pusherError && $code >= 4100 && $code <= 4199) {
            return Observable::timer(1001);
        }

        // Errors 4200-4299 connection closed by Pusher, reconnect immediately, we wait 0.001 second
        if ($pusherError && $code >= 4200 && $code <= 4299) {
            return Observable::timer(1);
        }

        // Double our delay each time we get here
        $this->delay *= 2;

        return Observable::timer($this->delay);
    }
}


1		<?php declare(strict_types=1);
2
3		namespace ApiClients\Client\Pusher;
4
5		use React\Dns\Resolver\Resolver;
6		use React\EventLoop\LoopInterface;
7		use RuntimeException;
8		use Rx\Observable;
9		use Rx\Scheduler;
10		use Rx\Subject\Subject;
11		use Rx\Websocket\WebsocketErrorException;
12		use Throwable;
13
14		final class AsyncClient
15		{
16		const DEFAULT_DELAY = 200;
17		const NO_ACTIVITY_TIMEOUT = 120;
18		const NO_PING_RESPONSE_TIMEOUT = 30;
19
20		/**
21		* @var Observable
22		*/
23		private $messages;
24
25		/**
26		* @var array
27		*/
28		private $channels = [];
29
30		/**
31		* @var int
32		*/
33		private $delay = self::DEFAULT_DELAY;
34
35		/**
36		* @var Subject
37		*/
38		private $client;
39
40		/**
41		* @var Observable
42		*/
43		private $connected;
44
45		/**
46		* @internal
47		* @param Subject $client
48		* @throws \InvalidArgumentException
49		*/
50	10	public function __construct(Subject $client)
51		{
52	10	$this->client = $client;
53
54		/** @var Observable $events */
55		$events = $client
56	10	->_ApiClients_jsonDecode()
57	10	->map([Event::class, 'createFromMessage'])
58	10	->singleInstance();
59
60		$pusherErrors = $events
61	10	->filter([Event::class, 'isError'])
62	10	->flatMap(function (Event $event) {
63		$throwable = new PusherErrorException($event->getData()['message'], (int)$event->getData()['code']);
64
65		return Observable::error($throwable);
66	10	});
67
68	10	$this->connected = $events
69	10	->filter([Event::class, 'connectionEstablished'])
70	10	->singleInstance();
71
72	10	$this->messages = $events
73	10	->merge($this->timeout($events))
74	10	->merge($pusherErrors)
75		->retryWhen(function (Observable $errors) {
76	10	return $errors->flatMap(function (Throwable $throwable) {
77	3	return $this->handleLowLevelError($throwable);
78	10	});
79	10	})
80	10	->singleInstance();
81	10	}
82
83		/**
84		* @param LoopInterface $loop
85		* @param string $app Application ID
86		* @param Resolver $resolver Optional DNS resolver
87		* @throws \InvalidArgumentException
88		* @return AsyncClient
89		*/
90		public static function create(LoopInterface $loop, string $app, Resolver $resolver = null): AsyncClient
91		{
92		try {
93		Scheduler::setDefaultFactory(function () use ($loop) {
94		return new Scheduler\EventLoopScheduler($loop);
		0 ignored issues – show Documentation introduced 2017-04-05 19:21 UTC by Report Bug Copy Issue Report `$loop` is of type `object<React\EventLoop\LoopInterface>`, but the function expects a `callable`. It seems like the type of the argument is not accepted by the function/method which you are calling. In some cases, in particular if PHP’s automatic type-juggling kicks in this might be fine. In other cases, however this might be a bug. We suggest to add an explicit type cast like in the following example: function acceptsInteger($int) { } $x = '123'; // string "123" // Instead of acceptsInteger($x); // we recommend to use acceptsInteger((integer) $x); Loading history...
95		});
96		} catch (Throwable $t) {
		0 ignored issues – show Coding Style Comprehensibility introduced 2017-04-05 19:21 UTC by Report Bug Copy Issue Report Consider adding a comment why this CATCH block is empty. Loading history...
97		}
98
99		return new self(
100		new Websocket(ApiSettings::createUrl($app), false, [], $loop, $resolver)
101		);
102		}
103
104		/**
105		* Listen on a channel.
106		*
107		* @param string $channel Channel to listen on
108		* @throws \InvalidArgumentException
109		* @return Observable
110		*/
111	10	public function channel(string $channel): Observable
112		{
113		// Only join a channel once
114	10	if (isset($this->channels[$channel])) {
115		return $this->channels[$channel];
116		}
117
118		// Ensure we only get messages for the given channel
119	10	$channelMessages = $this->messages->filter(function (Event $event) use ($channel) {
120	6	return $event->getChannel() !== '' && $event->getChannel() === $channel;
121	10	});
122
123	10	$subscribe = $this->connected
124	10	->do(function () use ($channel) {
125		// Subscribe to pusher channel after connected
126	6	$this->send(Event::subscribeOn($channel));
127	10	})
128	10	->flatMapTo(Observable::empty());
129
130		// Observable representing channel events
131		$events = $channelMessages
132	10	->merge($subscribe)
133	10	->filter([Event::class, 'subscriptionSucceeded'])
134	10	->finally(function () use ($channel) {
135		// Send unsubscribe event
136	10	$this->send(Event::unsubscribeOn($channel));
137
138		// Remove our channel from the channel list so we don't resubscribe in case we reconnect
139	10	unset($this->channels[$channel]);
140	10	});
141
142	10	$this->channels[$channel] = $events;
143
144	10	return $this->channels[$channel];
145		}
146
147		/**
148		* Send a message through the client.
149		*
150		* @param array $message Message to send, will be json encoded
151		*
152		*/
153	10	public function send(array $message)
154		{
155	10	$this->client->onNext(json_encode($message));
156	10	}
157
158		/**
159		* Returns an observable of TimeoutException.
160		* The timeout observable will get cancelled every time a new event is received.
161		*
162		* @param Observable $events
163		* @return Observable
164		*/
165		public function timeout(Observable $events): Observable
166		{
167	10	$timeoutDuration = $this->connected->map(function (Event $event) {
168	6	return ($event->getData()['activity_timeout'] ?? self::NO_ACTIVITY_TIMEOUT) * 1000;
169	10	});
170
171		return $timeoutDuration
172	10	->combineLatest([$events])
173	10	->pluck(0)
174	10	->concat(Observable::of(-1))
175	10	->flatMapLatest(function (int $time) {
176
177		// If the events observable ends, return an empty observable so we don't keep the stream alive
178	7	if ($time === -1) {
179	4	return Observable::empty();
180		}
181
182	6	return Observable::never()
183	6	->timeout($time)
184	6	->catch(function () use ($time) {
185		// ping (do something that causes incoming stream to get a message)
186	2	$this->send(Event::ping());
187		// this timeout will actually timeout with a TimeoutException - causing
188		// everything above this to dispose
189	2	return Observable::never()->timeout($time);
190	6	});
191	10	});
192		}
193
194		/**
195		* Handle errors as described at https://pusher.com/docs/pusher_protocol#error-codes.
196		* @param Throwable $throwable
197		* @return Observable
198		*/
199	3	private function handleLowLevelError(Throwable $throwable): Observable
200		{
201		// Only allow certain, relevant, exceptions
202	3	if (!($throwable instanceof WebsocketErrorException) &&
203	3	!($throwable instanceof RuntimeException) &&
204	3	!($throwable instanceof PusherErrorException)
205		) {
206	2	return Observable::error($throwable);
207		}
208
209	1	$code = $throwable->getCode();
210	1	$pusherError = ($throwable instanceof WebsocketErrorException \|\| $throwable instanceof PusherErrorException);
211
212		// Errors 4000-4099, don't retry connecting
213	1	if ($pusherError && $code >= 4000 && $code <= 4099) {
214		return Observable::error($throwable);
215		}
216
217		// Errors 4100-4199 reconnect after 1 or more seconds, we do it after 1.001 second
218	1	if ($pusherError && $code >= 4100 && $code <= 4199) {
219		return Observable::timer(1001);
220		}
221
222		// Errors 4200-4299 connection closed by Pusher, reconnect immediately, we wait 0.001 second
223	1	if ($pusherError && $code >= 4200 && $code <= 4299) {
224		return Observable::timer(1);
225		}
226
227		// Double our delay each time we get here
228	1	$this->delay *= 2;
229
230	1	return Observable::timer($this->delay);
231		}
232		}
233

php-api-clients / pusher

GitHub Access Token became invalid

Pull Request — master (#7)

AsyncClient::send() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like