Lines 40.67% 24 / 59
Functions and Methods 17.85% 5 / 28
Classes and Traits 0.00% 0 / 1
Name Lines Functions and Methods CRAP Classes and Traits
FakeWebsocketConnection 40.67% 24 / 59 17.85% 5 / 28 391.93 0.00% 0 / 1
 __construct 100.00% 1 / 1 100.00% 1 / 1 1
 queueFrame 100.00% 2 / 2 100.00% 1 / 1 1
 replyWhenSent 100.00% 1 / 1 100.00% 1 / 1 1
 answerEverySend 100.00% 1 / 1 100.00% 1 / 1 1
 holdSendsUntil 0.00% 0 / 1 0.00% 0 / 1 2
 holdCloseUntil 0.00% 0 / 1 0.00% 0 / 1 2
 listSentOfType 0.00% 0 / 1 0.00% 0 / 1 2
 receive 28.57% 2 / 7 0.00% 0 / 1 6.28
 sendText 92.85% 13 / 14 0.00% 0 / 1 8.02
 close 0.00% 0 / 8 0.00% 0 / 1 12
 isClosed 0.00% 0 / 1 0.00% 0 / 1 2
 getCloseInfo 0.00% 0 / 1 0.00% 0 / 1 2
 onClose 0.00% 0 / 1 0.00% 0 / 1 2
 getId 0.00% 0 / 1 0.00% 0 / 1 2
 getLocalAddress 0.00% 0 / 1 0.00% 0 / 1 2
 getRemoteAddress 0.00% 0 / 1 0.00% 0 / 1 2
 getTlsInfo 0.00% 0 / 1 0.00% 0 / 1 2
 isCompressionEnabled 0.00% 0 / 1 0.00% 0 / 1 2
 sendBinary 0.00% 0 / 1 0.00% 0 / 1 2
 streamText 0.00% 0 / 1 0.00% 0 / 1 2
 streamBinary 0.00% 0 / 1 0.00% 0 / 1 2
 ping 0.00% 0 / 1 0.00% 0 / 1 2
 getCount 0.00% 0 / 1 0.00% 0 / 1 2
 getTimestamp 0.00% 0 / 1 0.00% 0 / 1 2
 getHandshakeResponse 0.00% 0 / 1 0.00% 0 / 1 2
 getIterator 0.00% 0 / 2 0.00% 0 / 1 6
 encodeFrame 100.00% 1 / 1 100.00% 1 / 1 1
 wakeReceiver 75.00% 3 / 4 0.00% 0 / 1 2.06
1<?php
2
3declare(strict_types=1);
4
5namespace Stewart\Testing\Websocket;
6
7use Amp\ByteStream\ReadableStream;
8use Amp\Cancellation;
9use Amp\DeferredFuture;
10use Amp\Http\Client\Response;
11use Amp\Socket\InternetAddress;
12use Amp\Socket\SocketAddress;
13use Amp\Socket\TlsInfo;
14use Amp\Websocket\Client\WebsocketConnection;
15use Amp\Websocket\WebsocketCloseCode;
16use Amp\Websocket\WebsocketCloseInfo;
17use Amp\Websocket\WebsocketCount;
18use Amp\Websocket\WebsocketMessage;
19use Amp\Websocket\WebsocketTimestamp;
20use Closure;
21use IteratorAggregate;
22use LogicException;
23use SplQueue;
24use Stewart\Testing\Async\Latch;
25use Traversable;
26
27/** @implements IteratorAggregate<int, WebsocketMessage> */
28final class FakeWebsocketConnection implements IteratorAggregate, WebsocketConnection
29{
30    /** @var list<array<string, mixed>> */
31    public array $sent = [];
32
33    /** @var SplQueue<string> */
34    private SplQueue $inbound;
35
36    /** @var list<DeferredFuture<null>> */
37    private array $waiting = [];
38
39    /** @var array<string, list<array<string, mixed>>> */
40    private array $onSend = [];
41
42    /** @var array<string, Closure(array<string, mixed>): ?array<string, mixed>> */
43    private array $answerers = [];
44
45    private bool $closed = false;
46
47    private ?WebsocketCloseInfo $closeInfo = null;
48
49    /** @var list<Closure(int, WebsocketCloseInfo): void> */
50    private array $onClose = [];
51
52    private ?Latch $sendRelease = null;
53
54    private ?Latch $closeRelease = null;
55
56    public function __construct()
57    {
58        $this->inbound = new SplQueue();
59    }
60
61    /** @param array<string, mixed> $frame */
62    public function queueFrame(array $frame): void
63    {
64        $this->inbound->enqueue(self::encodeFrame($frame));
65        $this->wakeReceiver();
66    }
67
68    /** @param array<string, mixed> $frame */
69    public function replyWhenSent(string $type, array $frame): void
70    {
71        $this->onSend[$type][] = $frame;
72    }
73
74    // The answer may be null to reply later through queueFrame().
75    /** @param Closure(array<string, mixed>): ?array<string, mixed> $answer */
76    public function answerEverySend(string $type, Closure $answer): void
77    {
78        $this->answerers[$type] = $answer;
79    }
80
81    public function holdSendsUntil(Latch $release): void
82    {
83        $this->sendRelease = $release;
84    }
85
86    public function holdCloseUntil(Latch $release): void
87    {
88        $this->closeRelease = $release;
89    }
90
91    /** @return list<array<string, mixed>> */
92    public function listSentOfType(string $type): array
93    {
94        return array_values(array_filter($this->sent, static fn(array $command): bool => ($command['type'] ?? null) === $type));
95    }
96
97    public function receive(?Cancellation $cancellation = null): ?WebsocketMessage
98    {
99        while ($this->inbound->isEmpty()) {
100            if ($this->closed) {
101                return null;
102            }
103
104            $deferred = new DeferredFuture();
105            $this->waiting[] = $deferred;
106            $deferred->getFuture()->await($cancellation);
107        }
108
109        return WebsocketMessage::fromText($this->inbound->dequeue());
110    }
111
112    public function sendText(string $data): void
113    {
114        $this->sendRelease?->waitUntilOpen();
115
116        if ($this->closed) {
117            throw new LogicException('The fake socket is closed.');
118        }
119
120        /** @var array<string, mixed> $command */
121        $command = json_decode($data, true, 512, \JSON_THROW_ON_ERROR);
122        $this->sent[] = $command;
123
124        $type = \is_string($command['type'] ?? null) ? $command['type'] : '';
125        $id = $command['id'] ?? null;
126
127        $frames = $this->onSend[$type] ?? [];
128        unset($this->onSend[$type]);
129
130        $answer = $frames === [] && isset($this->answerers[$type]) ? ($this->answerers[$type])($command) : null;
131
132        if ($answer !== null) {
133            $frames = [$answer];
134        }
135
136        foreach ($frames as $frame) {
137            $this->queueFrame(\is_int($id) ? ['id' => $id, ...$frame] : $frame);
138        }
139    }
140
141    public function close(int $code = WebsocketCloseCode::NORMAL_CLOSE, string $reason = ''): void
142    {
143        if ($this->closed) {
144            return;
145        }
146
147        $this->closed = true;
148        $closeInfo = $this->closeInfo = new WebsocketCloseInfo($code, $reason, microtime(true), false);
149        $this->wakeReceiver();
150        $this->closeRelease?->waitUntilOpen();
151
152        foreach ($this->onClose as $handler) {
153            $handler($this->getId(), $closeInfo);
154        }
155    }
156
157    public function isClosed(): bool
158    {
159        return $this->closed;
160    }
161
162    public function getCloseInfo(): WebsocketCloseInfo
163    {
164        return $this->closeInfo ??= new WebsocketCloseInfo(WebsocketCloseCode::NORMAL_CLOSE, '', microtime(true), false);
165    }
166
167    public function onClose(Closure $onClose): void
168    {
169        $this->onClose[] = $onClose;
170    }
171
172    public function getId(): int
173    {
174        return 1;
175    }
176
177    public function getLocalAddress(): SocketAddress
178    {
179        return new InternetAddress('127.0.0.1', 1234);
180    }
181
182    public function getRemoteAddress(): SocketAddress
183    {
184        return new InternetAddress('127.0.0.1', 8123);
185    }
186
187    public function getTlsInfo(): ?TlsInfo
188    {
189        return null;
190    }
191
192    public function isCompressionEnabled(): bool
193    {
194        return false;
195    }
196
197    public function sendBinary(string $data): void
198    {
199        throw new LogicException('Stewart never sends binary frames.');
200    }
201
202    public function streamText(ReadableStream $stream): void
203    {
204        throw new LogicException('Stewart never streams frames.');
205    }
206
207    public function streamBinary(ReadableStream $stream): void
208    {
209        throw new LogicException('Stewart never streams frames.');
210    }
211
212    public function ping(): void {}
213
214    public function getCount(WebsocketCount $type): int
215    {
216        return 0;
217    }
218
219    public function getTimestamp(WebsocketTimestamp $type): float
220    {
221        return microtime(true);
222    }
223
224    public function getHandshakeResponse(): Response
225    {
226        throw new LogicException('The fake socket has no handshake response.');
227    }
228
229    public function getIterator(): Traversable
230    {
231        while (($message = $this->receive()) !== null) {
232            yield $message;
233        }
234    }
235
236    /** @param array<string, mixed> $frame */
237    private static function encodeFrame(array $frame): string
238    {
239        return json_encode($frame, \JSON_THROW_ON_ERROR);
240    }
241
242    private function wakeReceiver(): void
243    {
244        $waiting = $this->waiting;
245        $this->waiting = [];
246
247        foreach ($waiting as $deferred) {
248            $deferred->complete();
249        }
250    }
251}