Code Coverage
 
Lines
Functions and Methods
Classes and Traits
Total
90.51%
143 / 158
75.00%
15 / 20
 CRAP
0.00%
0 / 1
SynchronousStreamSession
90.51%
143 / 158
75.00%
15 / 20
 66.40
0.00%
0 / 1
 __construct
100.00%
5 / 5
100.00%
1 / 1
 1
 open
88.89%
24 / 27
0.00%
0 / 1
 5.03
 write
100.00%
1 / 1
100.00%
1 / 1
 1
 sendRequest
100.00%
1 / 1
100.00%
1 / 1
 1
 send
100.00%
13 / 13
100.00%
1 / 1
 4
 readResponse
100.00%
1 / 1
100.00%
1 / 1
 1
 readResponseDiscardingBody
100.00%
1 / 1
100.00%
1 / 1
 1
 receive
100.00%
25 / 25
100.00%
1 / 1
 6
 release
100.00%
1 / 1
100.00%
1 / 1
 1
 close
100.00%
4 / 4
100.00%
1 / 1
 2
 assertActive
100.00%
2 / 2
100.00%
1 / 1
 2
 connectTimeout
100.00%
3 / 3
100.00%
1 / 1
 2
 look
50.00%
2 / 4
0.00%
0 / 1
 2.50
 lookAt
100.00%
3 / 3
100.00%
1 / 1
 3
 splitTimeout
100.00%
5 / 5
100.00%
1 / 1
 4
 setBlocking
25.00%
1 / 4
0.00%
0 / 1
 6.80
 writeAll
100.00%
23 / 23
100.00%
1 / 1
 7
 awaitWritable
66.67%
8 / 12
0.00%
0 / 1
 4.59
 readChunk
86.36%
19 / 22
0.00%
0 / 1
 12.37
 timeout
100.00%
1 / 1
100.00%
1 / 1
 1
1<?php
2
3/**
4 * SPDX-License-Identifier: EUPL-1.2
5 *
6 * This file is part of icap-flow.
7 *
8 * Licensed under the EUPL, Version 1.2 only (the "Licence");
9 * you may not use this work except in compliance with the Licence.
10 * You may obtain a copy of the Licence at:
11 *
12 *     https://joinup.ec.europa.eu/collection/eupl/eupl-text-eupl-12
13 *
14 * Unless required by applicable law or agreed to in writing, software
15 * distributed under the Licence is distributed on an "AS IS" basis,
16 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
17 */
18
19declare(strict_types=1);
20
21namespace Ndrstmr\Icap\Transport;
22
23use Amp\Cancellation;
24use Ndrstmr\Icap\Config;
25use Ndrstmr\Icap\DTO\IcapResponse;
26use Ndrstmr\Icap\Exception\IcapConnectionException;
27use Ndrstmr\Icap\Exception\IcapTimeoutException;
28use Ndrstmr\Icap\IcapUri;
29use Ndrstmr\Icap\OperationDeadline;
30
31/**
32 * One TCP connection of the {@see SynchronousStreamTransport}, for as many write/read round trips as an exchange
33 * needs: the preview and the rest of the body of RFC 3507 ยง4.5 travel on it as one request. Calls block, and there is
34 * no TLS; use {@see AsyncAmpTransport} for icaps://.
35 *
36 * A request goes out through a non-blocking socket that is waited for with stream_select(): a blocking send() waits for
37 * ever on macOS for a peer that has stopped reading. A chunk is sent completely or the call fails. A {@see write()} has
38 * one window of `streamTimeout` for all its chunks, the time its producer takes included, as in
39 * {@see AmpTransportSession}; the request of {@see SynchronousStreamTransport::request()}, which has the connection to
40 * itself, keeps the limit of that transport ({@see sendRequest()}). A response is read in blocking mode in the two
41 * windows of {@see AmpTransportSession}: `responseStartTimeout` for the server to begin, then `streamTimeout` from its
42 * first byte as a maximum for the response to arrive in full.
43 *
44 * A native wait cannot be interrupted by the caller's cancellation, which in an operation of the client is also its end
45 * ({@see OperationDeadline}). Each wait is bounded by the time that is left of the operation and of its own limit, and
46 * both are looked at between the waits: before the producer is asked for a first chunk, whenever it hands one back,
47 * after every write and every wait before the result counts, and after every read before its bytes, its end of the
48 * stream or its failure are told. A producer that does not return, or a call that PHP cannot interrupt, is noticed when
49 * control comes back. A call that does not finish closes the session for good; a limit that passes is an
50 * {@see IcapTimeoutException}, a cancellation of the caller stays what it is and comes first.
51 *
52 * A read that brings in bytes behind the end of a response fails, as in {@see AmpTransportSession}: they answer no
53 * request.
54 */
55final class SynchronousStreamSession implements DiscardingTransportSession
56{
57    private const int READ_CHUNK_SIZE = 8192;
58
59    /** A chunk is written in slices of this size, so that a large one is not copied again for every short write. */
60    private const int WRITE_SLICE_SIZE = 65536;
61
62    private bool $disposed = false;
63
64    private readonly ResponseFrameReader $reader;
65
66    /**
67     * @param resource $stream
68     */
69    private function __construct(
70        private readonly Config $config,
71        private $stream,
72        private readonly ?Cancellation $cancellation,
73    ) {
74        $this->reader = new ResponseFrameReader(
75            maxResponseSize: $config->getMaxResponseSize(),
76            maxHeaderLineLength: $config->getMaxHeaderLineLength(),
77            maxHeaderCount: $config->getMaxHeaderCount(),
78        );
79    }
80
81    /**
82     * @throws IcapConnectionException When a TLS context is configured, or the connection fails
83     */
84    public static function open(Config $config, ?Cancellation $cancellation): self
85    {
86        if ($config->getTlsContext() !== null) {
87            throw new IcapConnectionException(
88                'SynchronousStreamTransport does not support TLS; use AsyncAmpTransport for icaps://.',
89            );
90        }
91
92        // The blocking PHP stream API does not allow interrupting a call in flight, so the cancellation is looked at
93        // between the calls, and every wait is bounded by a timeout of its own and by what is left of the operation.
94        self::look($cancellation);
95
96        $errno = 0;
97        $errstr = '';
98        $address = IcapUri::tcp($config->host, $config->port);
99        $stream = @stream_socket_client(
100            $address,
101            $errno,
102            $errstr,
103            self::connectTimeout($config, $cancellation),
104            STREAM_CLIENT_CONNECT,
105        );
106        if ($stream === false) {
107            // The end of the operation may be what cut the attempt short (see connectTimeout()).
108            self::look($cancellation);
109            throw new IcapConnectionException(
110                sprintf('Connection to %s failed: %s', $address, $errstr !== '' ? $errstr : 'unknown error'),
111            );
112        }
113
114        try {
115            // A cancellation that came in while connecting must not reach the producer.
116            self::look($cancellation);
117
118            [$seconds, $microseconds] = self::splitTimeout($config->getStreamTimeout());
119            stream_set_timeout($stream, $seconds, $microseconds);
120        } catch (\Throwable $e) {
121            fclose($stream);
122            throw $e;
123        }
124
125        return new self($config, $stream, $cancellation);
126    }
127
128    /**
129     * All chunks within one window of `streamTimeout`, the time the producer takes between them and after the last one
130     * included ({@see TransportSession::write()}).
131     */
132    #[\Override]
133    public function write(iterable $chunks): void
134    {
135        $this->send($chunks, hrtime(true) + $this->config->getStreamTimeout() * 1e9);
136    }
137
138    /**
139     * The request of {@see SynchronousStreamTransport::request()}, which has the connection to itself, with the limit
140     * of that transport: `streamTimeout` is the time the peer may go without taking a byte, and every byte it takes
141     * starts it again. The end of the operation bounds the total.
142     *
143     * @internal
144     *
145     * @param iterable<string> $chunks
146     */
147    public function sendRequest(iterable $chunks): void
148    {
149        $this->send($chunks, null);
150    }
151
152    /**
153     * @param iterable<string> $chunks
154     * @param float|null       $windowEndsAt hrtime(true) at which the window of the call is over; null for the limit
155     *                                       of {@see sendRequest()}
156     */
157    private function send(iterable $chunks, ?float $windowEndsAt): void
158    {
159        $this->assertActive();
160
161        try {
162            $timeout = $this->config->getStreamTimeout();
163            // Cancelled or out of time before it began: the producer is not asked for anything.
164            self::lookAt($this->cancellation, $windowEndsAt, $timeout);
165            // A blocking send() honours the stream timeout on Linux only, a blocking read everywhere: the request goes
166            // out through a non-blocking socket, the response comes in through a blocking one.
167            self::setBlocking($this->stream, false);
168            foreach ($chunks as $chunk) {
169                // The producer has handed control back. writeAll() looks after every write, so the producer is asked
170                // for the next chunk only while the call may go on.
171                self::lookAt($this->cancellation, $windowEndsAt, $timeout);
172                if ($chunk !== '') {
173                    self::writeAll($this->stream, $chunk, $this->cancellation, $timeout, $windowEndsAt);
174                }
175            }
176            // The producer may have run on after the last chunk went out.
177            self::lookAt($this->cancellation, $windowEndsAt, $timeout);
178            self::setBlocking($this->stream, true);
179        } catch (\Throwable $e) {
180            $this->close();
181            throw $e;
182        }
183    }
184
185    #[\Override]
186    public function readResponse(): string
187    {
188        return $this->receive(static fn (ResponseFrameReader $reader, \Closure $produce): string => $reader->readFrom($produce));
189    }
190
191    #[\Override]
192    public function readResponseDiscardingBody(int $maxResponseBytes): IcapResponse
193    {
194        return $this->receive(static fn (ResponseFrameReader $reader, \Closure $produce): IcapResponse => $reader->readDiscardingBody($produce, $maxResponseBytes));
195    }
196
197    /**
198     * One response, in two limits one after the other: the server's time to begin its answer, then a maximum for the
199     * answer to arrive in full. Neither starts again with a byte; each read waits for what is left.
200     *
201     * @template T
202     *
203     * @param \Closure(ResponseFrameReader, \Closure(): ?string): T $consume
204     *
205     * @return T
206     */
207    private function receive(\Closure $consume): mixed
208    {
209        $this->assertActive();
210
211        try {
212            $stream = $this->stream;
213            $cancellation = $this->cancellation;
214            $timeout = $this->config->getStreamTimeout();
215            $begun = false;
216            $limit = $this->config->getResponseStartTimeout();
217            $phaseEnds = hrtime(true) + $limit * 1e9;
218            $response = $consume(
219                $this->reader,
220                static function () use ($stream, $cancellation, $timeout, &$limit, &$phaseEnds, &$begun): ?string {
221                    $bytes = self::readChunk($stream, $cancellation, $limit, $phaseEnds, $begun ? 'Reading the response' : 'Waiting for the response to begin');
222                    if ($bytes !== null && !$begun) {
223                        $begun = true;
224                        $limit = $timeout;
225                        $phaseEnds = hrtime(true) + $limit * 1e9;
226                    }
227
228                    return $bytes;
229                },
230            );
231
232            // Every read waited for what was left of its phase and was looked at when it returned; what is left to look
233            // at is the time between the last of them and the complete response, which decoding it takes. A caller who
234            // cancelled in it comes first, whether or not the transfer ran out as well.
235            self::look($cancellation);
236            if (hrtime(true) >= $phaseEnds) {
237                // A response is complete only after its first byte: the transfer's window is the one that ran out.
238                throw self::timeout('Reading the response', $limit);
239            }
240
241            return $response;
242        } catch (\Throwable $e) {
243            $this->close();
244            throw $e;
245        }
246    }
247
248    /**
249     * There is no pool: the connection ends with the exchange.
250     */
251    #[\Override]
252    public function release(): void
253    {
254        $this->close();
255    }
256
257    #[\Override]
258    public function close(): void
259    {
260        if ($this->disposed) {
261            return;
262        }
263        $this->disposed = true;
264        fclose($this->stream);
265    }
266
267    private function assertActive(): void
268    {
269        if ($this->disposed) {
270            throw new \LogicException('TransportSession has already been released or closed.');
271        }
272    }
273
274    /**
275     * How long the connection may take: `socketTimeout`, or what is left of the operation if that is less, and a
276     * hair more (10 ms), so that a connect which the end of the operation cuts short fails after that end and is
277     * told as it, an {@see IcapTimeoutException}, not as a failed connection; the 10 ms are also what a PHP
278     * timeout of 0 would otherwise be.
279     */
280    private static function connectTimeout(Config $config, ?Cancellation $cancellation): float
281    {
282        return $cancellation instanceof OperationDeadline
283            ? min($config->getSocketTimeout(), $cancellation->remaining() + 0.01)
284            : $config->getSocketTimeout();
285    }
286
287    /**
288     * Looks at the cancellation: the caller's, as the exception it is, and, for an operation of the client, the end
289     * of the operation, as an {@see IcapTimeoutException}.
290     */
291    private static function look(?Cancellation $cancellation): void
292    {
293        if ($cancellation instanceof OperationDeadline) {
294            $cancellation->checkpoint();
295
296            return;
297        }
298
299        $cancellation?->throwIfRequested();
300    }
301
302    /**
303     * Looks at the cancellation, then at the window of a write if it has one.
304     */
305    private static function lookAt(?Cancellation $cancellation, ?float $windowEndsAt, float $timeout): void
306    {
307        self::look($cancellation);
308        if ($windowEndsAt !== null && hrtime(true) >= $windowEndsAt) {
309            throw self::timeout('Writing the request', $timeout);
310        }
311    }
312
313    /**
314     * PHP takes the stream timeout as whole seconds and microseconds. A value
315     * that is not a positive number gives up at once: handed on as it is, a
316     * negative one could become a tv_sec of -1, which PHP reads as "wait for
317     * ever".
318     *
319     * @return array{int, int}
320     */
321    private static function splitTimeout(float $timeout): array
322    {
323        if (!is_finite($timeout) || $timeout <= 0.0) {
324            return [0, 0];
325        }
326
327        $seconds = (int) floor($timeout);
328        $microseconds = (int) round(($timeout - $seconds) * 1_000_000);
329
330        return $microseconds < 1_000_000 ? [$seconds, $microseconds] : [$seconds + 1, 0];
331    }
332
333    /**
334     * @param resource $stream
335     */
336    private static function setBlocking($stream, bool $blocking): void
337    {
338        if (!stream_set_blocking($stream, $blocking)) {
339            throw new IcapConnectionException(
340                sprintf('Switching the socket to %s mode failed.', $blocking ? 'blocking' : 'non-blocking'),
341            );
342        }
343    }
344
345    /**
346     * Sends one chunk completely or fails, through a non-blocking socket. A
347     * write may take less than it is given and then carries on where it
348     * stopped; one that takes nothing means the send buffer is full, and the
349     * call waits for the peer to read: until the end of the window of the
350     * call ($windowEndsAt, hrtime(true)), or without a window for one timeout
351     * without any progress at most. false ends the call at once.
352     *
353     * @param resource $stream
354     */
355    private static function writeAll($stream, string $chunk, ?Cancellation $cancellation, float $timeout, ?float $windowEndsAt = null): void
356    {
357        [$seconds, $microseconds] = self::splitTimeout($timeout);
358        $stall = ($seconds + $microseconds / 1_000_000) * 1e9;
359        $length = strlen($chunk);
360        $offset = 0;
361        $endsAt = $windowEndsAt;
362        while (true) {
363            // PHP raises a notice for every send that fails; the failure is reported below.
364            $written = @fwrite(
365                $stream,
366                $offset === 0 && $length <= self::WRITE_SLICE_SIZE
367                    ? $chunk
368                    : substr($chunk, $offset, self::WRITE_SLICE_SIZE),
369            );
370            // What the write did, a failure too, counts only if the caller has not cancelled and neither the operation
371            // nor the window ran out meanwhile.
372            self::lookAt($cancellation, $windowEndsAt, $timeout);
373            if ($written === false) {
374                throw new IcapConnectionException('Writing the request failed.');
375            }
376
377            $offset += $written;
378            if ($offset >= $length) {
379                return;
380            }
381
382            if ($written > 0) {
383                // Progress starts the time without progress again, not the window.
384                $endsAt = $windowEndsAt;
385            } else {
386                $endsAt ??= hrtime(true) + $stall;
387                self::awaitWritable($stream, $endsAt, $timeout, $cancellation);
388                // The rest of the chunk goes out only if the caller still wants it, and the operation and the window
389                // still have time.
390                self::lookAt($cancellation, $windowEndsAt, $timeout);
391            }
392        }
393    }
394
395    /**
396     * Waits until the socket may take more, or throws once $endsAt has passed.
397     * The wait is no longer than the time that is left of the operation. A
398     * wait that stream_select() cannot do (a signal interrupted it, or the
399     * descriptor lies beyond FD_SETSIZE) is no failure of the request: it
400     * returns after a moment and the caller tries to write again.
401     *
402     * @param resource $stream
403     * @param float    $endsAt hrtime(true) at which the wait is over
404     */
405    private static function awaitWritable($stream, float $endsAt, float $timeout, ?Cancellation $cancellation = null): void
406    {
407        $left = ($endsAt - hrtime(true)) / 1e9;
408        if ($cancellation instanceof OperationDeadline) {
409            $left = min($left, $cancellation->remaining());
410        }
411        if ($left <= 0.0) {
412            // The end of the operation, if that is what ran out, is told as such.
413            self::look($cancellation);
414            throw self::timeout('Writing the request', $timeout);
415        }
416
417        $read = null;
418        $write = [$stream];
419        $except = null;
420        [$seconds, $microseconds] = self::splitTimeout($left);
421        if (@stream_select($read, $write, $except, $seconds, $microseconds) === false) {
422            usleep(1_000);
423        }
424    }
425
426    /**
427     * One read of the response, null at the end of the stream. A read that
428     * timed out is no end of the stream (the frame reader would call it a
429     * connection closed too early) but a timeout. PHP reports one as false
430     * with `timed_out` set; a zero timeout answers with an empty string while
431     * the stream has not ended.
432     *
433     * With $phaseEndsAt (hrtime(true) at which the phase is over) the read waits for what is left of the phase,
434     * not for a whole timeout: a maximum, which no byte starts again. It waits no longer than the time that is
435     * left of the operation either.
436     *
437     * @param resource $stream
438     */
439    private static function readChunk($stream, ?Cancellation $cancellation, float $timeout, ?float $phaseEndsAt = null, string $what = 'Reading the response'): ?string
440    {
441        self::look($cancellation);
442
443        if ($phaseEndsAt !== null) {
444            $left = ($phaseEndsAt - hrtime(true)) / 1e9;
445            if ($cancellation instanceof OperationDeadline) {
446                $left = min($left, $cancellation->remaining());
447            }
448            if ($left <= 0.0) {
449                self::look($cancellation);
450                throw self::timeout($what, $timeout);
451            }
452            [$seconds, $microseconds] = self::splitTimeout($left);
453            stream_set_timeout($stream, $seconds, $microseconds);
454        }
455
456        $read = fread($stream, self::READ_CHUNK_SIZE);
457
458        // What the read brought - bytes, the end of the stream, a failure - counts only if the caller did not cancel and
459        // neither the operation nor the phase ran out while it blocked: a peer that hung up after that is no framing
460        // error, and bytes that came late are no answer.
461        self::look($cancellation);
462        if ($phaseEndsAt !== null && hrtime(true) >= $phaseEndsAt) {
463            throw self::timeout($what, $timeout);
464        }
465
466        if ($read !== false && $read !== '') {
467            return $read;
468        }
469
470        $meta = stream_get_meta_data($stream);
471        if ($meta['timed_out'] || ($read === '' && !$meta['eof'])) {
472            throw self::timeout($what, $timeout);
473        }
474        if ($read === false) {
475            throw new IcapConnectionException('Reading the response failed.');
476        }
477
478        return null;
479    }
480
481    private static function timeout(string $what, float $timeout): IcapTimeoutException
482    {
483        return new IcapTimeoutException(sprintf('%s timed out after %s seconds.', $what, $timeout));
484    }
485}