1 | <?php declare(strict_types=1); |
||
15 | final class AsyncClient |
||
16 | { |
||
17 | const DEFAULT_DELAY = 200; |
||
18 | const NO_ACTIVITY_TIMEOUT = 120; |
||
19 | const NO_PING_RESPONSE_TIMEOUT = 30; |
||
20 | |||
21 | /** |
||
22 | * @var Observable |
||
23 | */ |
||
24 | private $messages; |
||
25 | |||
26 | /** |
||
27 | * @var array |
||
28 | */ |
||
29 | private $channels = []; |
||
30 | |||
31 | /** |
||
32 | * @var int |
||
33 | */ |
||
34 | private $delay = self::DEFAULT_DELAY; |
||
35 | |||
36 | /** |
||
37 | * @var Subject |
||
38 | */ |
||
39 | private $client; |
||
40 | |||
41 | /** |
||
42 | * @var Observable |
||
43 | */ |
||
44 | private $connected; |
||
45 | |||
46 | /** |
||
47 | * @internal |
||
48 | * @param Subject $client |
||
49 | * @throws \InvalidArgumentException |
||
50 | */ |
||
51 | 19 | public function __construct(Subject $client) |
|
52 | { |
||
53 | 19 | $this->client = $client; |
|
54 | |||
55 | /** @var Observable $events */ |
||
56 | $events = $client |
||
57 | 19 | ->map(p('json_decode', true)) |
|
58 | 19 | ->map([Event::class, 'createFromMessage']) |
|
59 | 19 | ->singleInstance(); |
|
60 | |||
61 | $pusherErrors = $events |
||
62 | 19 | ->filter([Event::class, 'isError']) |
|
63 | 19 | ->flatMap(function (Event $event) { |
|
64 | $throwable = new PusherErrorException($event->getData()['message'], (int)$event->getData()['code']); |
||
65 | |||
66 | return Observable::error($throwable); |
||
67 | 19 | }); |
|
68 | |||
69 | 19 | $this->connected = $events |
|
70 | 19 | ->filter([Event::class, 'connectionEstablished']) |
|
71 | 19 | ->singleInstance(); |
|
72 | |||
73 | 19 | $this->messages = $events |
|
74 | 19 | ->merge($this->timeout($events)) |
|
75 | 19 | ->merge($pusherErrors) |
|
76 | 19 | ->singleInstance(); |
|
77 | 19 | } |
|
78 | |||
79 | /** |
||
80 | * @param LoopInterface $loop |
||
81 | * @param string $app Application ID |
||
82 | * @param Resolver $resolver Optional DNS resolver |
||
83 | * @throws \InvalidArgumentException |
||
84 | * @return AsyncClient |
||
85 | */ |
||
86 | public static function create(LoopInterface $loop, string $app, Resolver $resolver = null): AsyncClient |
||
87 | { |
||
88 | try { |
||
89 | 4 | Scheduler::setDefaultFactory(function () use ($loop) { |
|
90 | 3 | return new Scheduler\EventLoopScheduler($loop); |
|
|
|||
91 | 4 | }); |
|
92 | 2 | } catch (Throwable $t) { |
|
93 | } |
||
94 | |||
95 | 4 | return new self( |
|
96 | 4 | WebSocket::createFactory(ApiSettings::createUrl($app), false, [], $loop, $resolver) |
|
97 | ); |
||
98 | } |
||
99 | |||
100 | /** |
||
101 | * Listen on a channel. |
||
102 | * |
||
103 | * @param string $channel Channel to listen on |
||
104 | * @throws \InvalidArgumentException |
||
105 | * @return Observable |
||
106 | */ |
||
107 | 16 | public function channel(string $channel): Observable |
|
108 | { |
||
109 | // Only join a channel once |
||
110 | 16 | if (isset($this->channels[$channel])) { |
|
111 | 1 | return $this->channels[$channel]; |
|
112 | } |
||
113 | |||
114 | // Ensure we only get messages for the given channel |
||
115 | 16 | $channelMessages = $this->messages->filter(function (Event $event) use ($channel) { |
|
116 | 11 | return $event->getChannel() !== '' && $event->getChannel() === $channel; |
|
117 | 16 | }); |
|
118 | |||
119 | 16 | $subscribe = $this->connected |
|
120 | 16 | ->do(function () use ($channel) { |
|
121 | // Subscribe to pusher channel after connected |
||
122 | 11 | $this->send(Event::subscribeOn($channel)); |
|
123 | 16 | }) |
|
124 | 16 | ->flatMapTo(Observable::empty()); |
|
125 | |||
126 | // Observable representing channel events |
||
127 | 16 | $this->channels[$channel] = $channelMessages |
|
128 | 16 | ->merge($subscribe) |
|
129 | 16 | ->filter([Event::class, 'subscriptionSucceeded']) |
|
130 | ->retryWhen(function (Observable $errors) { |
||
131 | 16 | return $errors->flatMap(function (Throwable $throwable) { |
|
132 | 8 | return $this->handleLowLevelError($throwable); |
|
133 | 16 | }); |
|
134 | 16 | }) |
|
135 | 16 | ->finally(function () use ($channel) { |
|
136 | // Send unsubscribe event |
||
137 | 14 | $this->send(Event::unsubscribeOn($channel)); |
|
138 | |||
139 | // Remove our channel from the channel list so we don't resubscribe in case we reconnect |
||
140 | 14 | unset($this->channels[$channel]); |
|
141 | 16 | }) |
|
142 | 16 | ->singleInstance(); |
|
143 | |||
144 | 16 | 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 | 14 | public function send(array $message) |
|
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 | private function timeout(Observable $events): Observable |
||
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 | 8 | private function handleLowLevelError(Throwable $throwable): Observable |
|
232 | } |
||
233 |
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: