AsyncClient::channel() - Code Metrics - Inspection of "Improvement handle disconnects and reconnect when..." - php-api-clients/pusher - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#3)

by Cees-Jan

created 2017-04-05 19:15 UTC

AsyncClient::channel() B

↳ Parent: AsyncClient

Complexity

Conditions	3
Paths	2

Size

Total Lines	34
Code Lines	21

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	0
CRAP Score	12

Importance

Changes

Metric	Value
dl	0
loc	34
ccs	0
cts	7
cp	0
rs	8.8571
c	0
b	0
f	0
cc	3
eloc	21
nc	2
nop	1
crap	12

<?php declare(strict_types=1);

namespace ApiClients\Client\Pusher;

use React\Dns\Resolver\Resolver;
use React\EventLoop\LoopInterface;
use Rx\Disposable\CallbackDisposable;
use Rx\Observable;
use Rx\ObserverInterface;
use Rx\Scheduler;
use Rx\Websocket\Client as WebsocketClient;
use Rx\Websocket\MessageSubject;
use Throwable;

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

    protected $noActivityTimeout = self::NO_ACTIVITY_TIMEOUT;

    /**
     * @var LoopInterface
     */
    protected $loop;

    /**
     * @var Observable\RefCountObservable
     */
    protected $client;

    /**
     * @var Observable\AnonymousObservable
     */
    protected $messages;

    /**
     * @var MessageSubject
     */
    protected $sendSubject;

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

    /**
     * @var int
     */
    protected $delay = 200;

    /**
     * @param LoopInterface $loop
     * @param string $app Application ID
     * @param Resolver $resolver Optional DNS resolver
     * @return AsyncClient
     */
    public static function create(LoopInterface $loop, string $app, Resolver $resolver = null): AsyncClient
    {
        try {
            Scheduler::setAsyncFactory(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(
            $loop,
            new WebsocketClient(
                ApiSettings::createUrl($app),
                false,
                [],
                $loop,
                $resolver
            )
        );
    }

    /**
     * @internal
     */
    public function __construct(LoopInterface $loop, WebsocketClient $client)
    {
        $this->loop = $loop;
        $this->messages = $client
            // Save this subject for sending stuff
            ->do(function (MessageSubject $ms) {
                $this->sendSubject = $ms;

                // Resubscribe to an channels we where subscribed to when disconnected
                foreach ($this->channels as $channel => $_) {
                    $this->subscribeOnChannel($channel);
                }
            })

            // Make sure if there is a disconnect or something
            // that we unset the sendSubject
            ->finally(function () {
                $this->sendSubject = null;
            })


            ->flatMap(function (MessageSubject $ms) {
                return $ms;
            })

            // This is the ping/timeout functionality
            ->flatMapLatest(function ($x) {
                // this Observable emits the current value immediately
                // if another value comes along, this all gets disposed (because we are using flatMapLatest)
                // before the timeouts start get triggered
                return Observable::never()
                    ->timeout($this->noActivityTimeout * 1000)
                    ->catch(function () use ($x) {
                        // ping (do something that causes incoming stream to get a message)
                        $this->send(['event' => 'pusher:ping']);
                        // this timeout will actually timeout with a TimeoutException - causing
                        //   everything above this to dispose
                        return Observable::never()->timeout(self::NO_PING_RESPONSE_TIMEOUT * 1000);
                    })
                    ->startWith($x);
            })

            // Handle connection level errors
            ->retryWhen(function (Observable $errors) {
                return $errors->flatMap(function (Throwable $throwable) {
                    return $this->handleLowLevelError($throwable);
                });
            })

            // Decode JSON
            ->_ApiClients_jsonDecode()

            // Deal with connection established messages
            ->map(function (array $message) {
                $event = Event::createFromMessage($message);

                if ($event->getEvent() === 'pusher:connection_established') {
                    $this->setActivityTimeout($event);
                }

                return $event;
            })
            ->share();
    }

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

        $channelMessages = $this->messages->filter(function (Event $event) use ($channel) {
            return $event->getChannel() !== '' && $event->getChannel() === $channel;
        });

        $events = Observable::create(function (
            ObserverInterface $observer
        ) use (
            $channel,
            $channelMessages
        ) {
            $subscription = $channelMessages
                ->filter(function (Event $event) {
                    return $event->getEvent() !== 'pusher_internal:subscription_succeeded';
                })
                ->subscribe($observer);
function acceptsInteger($int) { }

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

// Instead of
acceptsInteger($x);

// we recommend to use
acceptsInteger((integer) $x);

            $this->subscribeOnChannel($channel);

            return new CallbackDisposable(function () use ($channel, $subscription) {
                $this->send(['event' => 'pusher:unsubscribe', 'data' => ['channel' => $channel]]);
                $subscription->dispose();
                unset($this->channels[$channel]);
            });
        });

        $this->channels[$channel] = $events->share();
        return $this->channels[$channel];
    }

    /**
     * Send a message through the client
     *
     * @param array $message Message to send, will be json encoded
     *
     * @return A bool indicating whether or not the connection was active
     *         and the given message has been pass onto the connection.
     */
    public function send(array $message): bool
    {
        if ($this->sendSubject ===  null) {
            return false;
        }

        $this->sendSubject->onNext(json_encode($message));
        return true;
    }

    private function handleLowLevelError(Throwable $throwable)

    {
        $this->delay *= 2;
        return Observable::timer($this->delay);
    }

    /**
     * @param string $channel
     */
    private function subscribeOnChannel(string $channel)
    {
        $this->send(['event' => 'pusher:subscribe', 'data' => ['channel' => $channel]]);
    }


    /**
     * Get connection activity timeout from connection established event
     *
     * @param Event $event
     */
    private function setActivityTimeout(Event $event)
    {
        $data = $event->getData();

        // No activity_timeout found on event
        if (!isset($data['activity_timeout'])) {
            return;
        }

        // activity_timeout holds zero or invalid value (we don't want to hammer Pusher)
        if ((int)$data['activity_timeout'] <= 0) {
            return;
        }

        $this->noActivityTimeout = (int)$data['activity_timeout'];
    }
}


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 Rx\Disposable\CallbackDisposable;
8		use Rx\Observable;
9		use Rx\ObserverInterface;
10		use Rx\Scheduler;
11		use Rx\Websocket\Client as WebsocketClient;
12		use Rx\Websocket\MessageSubject;
13		use Throwable;
14
15		final class AsyncClient
16		{
17		const NO_ACTIVITY_TIMEOUT = 120;
18		const NO_PING_RESPONSE_TIMEOUT = 30;
19
20		protected $noActivityTimeout = self::NO_ACTIVITY_TIMEOUT;
21
22		/**
23		* @var LoopInterface
24		*/
25		protected $loop;
26
27		/**
28		* @var Observable\RefCountObservable
29		*/
30		protected $client;
31
32		/**
33		* @var Observable\AnonymousObservable
34		*/
35		protected $messages;
36
37		/**
38		* @var MessageSubject
39		*/
40		protected $sendSubject;
41
42		/**
43		* @var array
44		*/
45		protected $channels = [];
46
47		/**
48		* @var int
49		*/
50		protected $delay = 200;
51
52		/**
53		* @param LoopInterface $loop
54		* @param string $app Application ID
55		* @param Resolver $resolver Optional DNS resolver
56		* @return AsyncClient
57		*/
58	1	public static function create(LoopInterface $loop, string $app, Resolver $resolver = null): AsyncClient
59		{
60		try {
61		Scheduler::setAsyncFactory(function () use ($loop) {
62		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...
63	1	});
64		} 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...
65		}
66
67	1	return new self(
68		$loop,
69	1	new WebsocketClient(
70	1	ApiSettings::createUrl($app),
71	1	false,
72	1	[],
73		$loop,
74		$resolver
75		)
76		);
77		}
78
79		/**
80		* @internal
81		*/
82		public function __construct(LoopInterface $loop, WebsocketClient $client)
83		{
84		$this->loop = $loop;
85		$this->messages = $client
86		// Save this subject for sending stuff
87		->do(function (MessageSubject $ms) {
88		$this->sendSubject = $ms;
89
90		// Resubscribe to an channels we where subscribed to when disconnected
91		foreach ($this->channels as $channel => $_) {
92		$this->subscribeOnChannel($channel);
93		}
94		})
95
96		// Make sure if there is a disconnect or something
97		// that we unset the sendSubject
98		->finally(function () {
99		$this->sendSubject = null;
100		})
101
102
103		->flatMap(function (MessageSubject $ms) {
104		return $ms;
105		})
106
107		// This is the ping/timeout functionality
108		->flatMapLatest(function ($x) {
109		// this Observable emits the current value immediately
110		// if another value comes along, this all gets disposed (because we are using flatMapLatest)
111		// before the timeouts start get triggered
112		return Observable::never()
113		->timeout($this->noActivityTimeout * 1000)
114		->catch(function () use ($x) {
115		// ping (do something that causes incoming stream to get a message)
116		$this->send(['event' => 'pusher:ping']);
117		// this timeout will actually timeout with a TimeoutException - causing
118		// everything above this to dispose
119		return Observable::never()->timeout(self::NO_PING_RESPONSE_TIMEOUT * 1000);
120		})
121		->startWith($x);
122		})
123
124		// Handle connection level errors
125		->retryWhen(function (Observable $errors) {
126		return $errors->flatMap(function (Throwable $throwable) {
127		return $this->handleLowLevelError($throwable);
128		});
129		})
130
131		// Decode JSON
132		->_ApiClients_jsonDecode()
133
134		// Deal with connection established messages
135		->map(function (array $message) {
136		$event = Event::createFromMessage($message);
137
138		if ($event->getEvent() === 'pusher:connection_established') {
139		$this->setActivityTimeout($event);
140		}
141
142		return $event;
143		})
144		->share();
145		}
146
147		/**
148		* Listen on a channel
149		*
150		* @param string $channel Channel to listen on
151		* @return Observable
152		*/
153		public function channel(string $channel): Observable
154		{
155		if (isset($this->channels[$channel])) {
156		return $this->channels[$channel];
157		}
158
159		$channelMessages = $this->messages->filter(function (Event $event) use ($channel) {
160		return $event->getChannel() !== '' && $event->getChannel() === $channel;
161		});
162
163		$events = Observable::create(function (
164		ObserverInterface $observer
165		) use (
166		$channel,
167		$channelMessages
168		) {
169		$subscription = $channelMessages
170		->filter(function (Event $event) {
171		return $event->getEvent() !== 'pusher_internal:subscription_succeeded';
172		})
173		->subscribe($observer);
		0 ignored issues – show Documentation introduced 2017-04-05 19:21 UTC by Report Bug Copy Issue Report `$observer` is of type `object<Rx\ObserverInterface>`, but the function expects a `callable\|null`. 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...
174
175		$this->subscribeOnChannel($channel);
176
177		return new CallbackDisposable(function () use ($channel, $subscription) {
178		$this->send(['event' => 'pusher:unsubscribe', 'data' => ['channel' => $channel]]);
179		$subscription->dispose();
180		unset($this->channels[$channel]);
181		});
182		});
183
184		$this->channels[$channel] = $events->share();
185		return $this->channels[$channel];
186		}
187
188		/**
189		* Send a message through the client
190		*
191		* @param array $message Message to send, will be json encoded
192		*
193		* @return A bool indicating whether or not the connection was active
194		* and the given message has been pass onto the connection.
195		*/
196		public function send(array $message): bool
197		{
198		if ($this->sendSubject === null) {
199		return false;
200		}
201
202		$this->sendSubject->onNext(json_encode($message));
203		return true;
204		}
205
206		private function handleLowLevelError(Throwable $throwable)
		0 ignored issues – show Unused Code introduced 2017-04-05 19:21 UTC by Report Bug Copy Issue Report The parameter `$throwable` is not used and could be removed. This check looks from parameters that have been defined for a function or method, but which are not used in the method body. Loading history...
207		{
208		$this->delay *= 2;
209		return Observable::timer($this->delay);
210		}
211
212		/**
213		* @param string $channel
214		*/
215		private function subscribeOnChannel(string $channel)
216		{
217		$this->send(['event' => 'pusher:subscribe', 'data' => ['channel' => $channel]]);
218		}
219
220
221		/**
222		* Get connection activity timeout from connection established event
223		*
224		* @param Event $event
225		*/
226		private function setActivityTimeout(Event $event)
227		{
228		$data = $event->getData();
229
230		// No activity_timeout found on event
231		if (!isset($data['activity_timeout'])) {
232		return;
233		}
234
235		// activity_timeout holds zero or invalid value (we don't want to hammer Pusher)
236		if ((int)$data['activity_timeout'] <= 0) {
237		return;
238		}
239
240		$this->noActivityTimeout = (int)$data['activity_timeout'];
241		}
242		}
243

php-api-clients / pusher

GitHub Access Token became invalid

Pull Request — master (#3)

AsyncClient::channel() B

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like