Code Coverage
 
Lines
Functions and Methods
Classes and Traits
Total
100.00%
56 / 56
100.00%
6 / 6
 CRAP
100.00%
1 / 1
AsyncAmpTransport
100.00%
56 / 56
100.00%
6 / 6
 14
100.00%
1 / 1
 __construct
100.00%
1 / 1
100.00%
1 / 1
 1
 request
100.00%
1 / 1
100.00%
1 / 1
 1
 requestDiscardingBody
100.00%
1 / 1
100.00%
1 / 1
 1
 exchange
100.00%
7 / 7
100.00%
1 / 1
 1
 openSession
100.00%
43 / 43
100.00%
1 / 1
 8
 acquireSocket
100.00%
3 / 3
100.00%
1 / 1
 2
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 Amp\CancelledException;
25use Amp\DeferredCancellation;
26use Amp\Socket\Socket as SocketInterface;
27use Ndrstmr\Icap\Config;
28use Ndrstmr\Icap\DTO\IcapResponse;
29use Ndrstmr\Icap\Exception\IcapTimeoutException;
30use Revolt\EventLoop;
31
32use function Amp\async;
33
34/**
35 * Asynchronous transport implementation using amphp/socket.
36 *
37 * Upgrades to TLS (icaps://) automatically when the supplied Config
38 * carries a {@see \Amp\Socket\ClientTlsContext}; otherwise connects
39 * plain tcp://. The response is bounded by Config::maxResponseSize to
40 * keep a hostile server from exhausting the client's memory.
41 *
42 * The optional user-supplied {@see Cancellation} is combined with the
43 * transport's own `streamTimeout` window for each phase (acquiring the
44 * socket, writing the request, reading the response); whichever comes
45 * first ends the phase. The caller's cancellation — also one that is a
46 * `TimeoutCancellation` — ends it with `Amp\CancelledException`, the
47 * transport's own window with {@see IcapTimeoutException}.
48 *
49 * Response framing is done by {@see ResponseFrameReader} so the read
50 * loop terminates as soon as the message is complete — no dependency
51 * on the server closing the socket.
52 *
53 * **Connection pooling (v2.1+).** When constructed with a
54 * {@see ConnectionPoolInterface}, the transport calls `acquire()`
55 * instead of opening a fresh TCP/TLS connection. A socket goes back
56 * with `release()` only when the caller of a {@see openSession()}
57 * session has validated the whole exchange; every other way out closes
58 * it. {@see request()} returns the raw response string, before any
59 * status or verdict has been checked, so it always closes the socket.
60 * Without a pool the transport opens a fresh socket per request and
61 * closes it after — identical to the v2.0 behaviour. A pool that limits
62 * its connections ({@see AmpConnectionPool}) limits the sessions and
63 * requests of this transport too: a session is a connection, and its room
64 * is back when the socket is closed or released. Without a pool the
65 * application limits its own concurrency.
66 *
67 * **Sessions (v2.1+).** Implements {@see SessionAwareTransport} so
68 * callers that need to send + receive multiple times on the same
69 * connection (the RFC 3507 §4.5 preview and its continuation) can call
70 * {@see openSession()}. A connection carries one outstanding request at
71 * a time (RFC 3507 §4.1).
72 */
73final class AsyncAmpTransport implements SessionAwareTransport, DiscardingTransport
74{
75    public function __construct(
76        private ?ConnectionPoolInterface $pool = null,
77    ) {
78    }
79
80    /**
81     * @param iterable<string> $rawRequest
82     * @return \Amp\Future<string>
83     */
84    #[\Override]
85    public function request(Config $config, iterable $rawRequest, ?Cancellation $cancellation = null): \Amp\Future
86    {
87        return $this->exchange($config, $rawRequest, $cancellation, static fn (AmpTransportSession $session): string => $session->readResponse());
88    }
89
90    /**
91     * @param iterable<string> $rawRequest
92     * @return \Amp\Future<IcapResponse>
93     */
94    #[\Override]
95    public function requestDiscardingBody(Config $config, iterable $rawRequest, int $maxResponseBytes, ?Cancellation $cancellation = null): \Amp\Future
96    {
97        return $this->exchange($config, $rawRequest, $cancellation, static fn (AmpTransportSession $session): IcapResponse => $session->readResponseDiscardingBody($maxResponseBytes));
98    }
99
100    /**
101     * One request on a session of its own, and what the caller reads from it.
102     *
103     * @template T
104     *
105     * @param iterable<string>                    $rawRequest
106     * @param \Closure(AmpTransportSession): T    $read
107     *
108     * @return \Amp\Future<T>
109     */
110    private function exchange(Config $config, iterable $rawRequest, ?Cancellation $cancellation, \Closure $read): \Amp\Future
111    {
112        /** @var \Amp\Future<T> $future */
113        $future = async(function () use ($config, $rawRequest, $cancellation, $read): mixed {
114            $session = $this->openSession($config, $cancellation);
115            try {
116                $session->write($rawRequest);
117
118                return $read($session);
119            } finally {
120                // The caller gets a response no one has validated yet: a
121                // 100 Continue or an error status are indistinguishable
122                // from a clean answer here. Reuse needs a validated
123                // exchange, which only a session offers.
124                $session->close();
125            }
126        });
127
128        return $future;
129    }
130
131    #[\Override]
132    public function openSession(Config $config, ?Cancellation $cancellation = null): AmpTransportSession
133    {
134        // Cancelled before it began: neither the pool nor the connector is asked.
135        $cancellation?->throwIfRequested();
136
137        // Waiting for the pool, connecting and the TLS handshake share one window: streamTimeout seconds, or the
138        // handshake limit if that is longer (the window must not cut it short). The end of the whole operation reaches
139        // the window as the cancellation of the caller and is shorter if it is. Timer and subscription are removed on
140        // every way out, not left to destructors: an exception that is still referenced keeps the arguments of its
141        // stack frames, and with them the cancellation, alive.
142        $setup = $config->getTlsContext() === null
143            ? $config->getStreamTimeout()
144            : max($config->getStreamTimeout(), $config->getTlsHandshakeTimeout());
145        $endsAt = hrtime(true) + $setup * 1e9;
146        $window = new DeferredCancellation();
147        $expired = false;
148        $deadline = EventLoop::delay($setup, function () use (&$expired, $window): void {
149            $expired = true;
150            $window->cancel();
151        });
152        $watch = $cancellation?->subscribe(static function () use ($window): void {
153            $window->cancel();
154        });
155        $socket = null;
156
157        try {
158            $socket = $this->acquireSocket($config, $window->getCancellation());
159
160            // A connector need not honour the token it was given, and one that holds the event
161            // loop leaves the timer no chance to fire. A socket it hands back after the caller
162            // gave up or the window passed is not used.
163            $cancellation?->throwIfRequested();
164            if (hrtime(true) >= $endsAt) {
165                $expired = true;
166                $window->cancel();
167            }
168            $window->getCancellation()->throwIfRequested();
169        } catch (\Throwable $e) {
170            $socket?->close();
171
172            if ($e instanceof CancelledException) {
173                // The caller's cancellation stays what it is; only the transport's own
174                // window passing becomes an IcapTimeoutException.
175                $cancellation?->throwIfRequested();
176                if ($expired) {
177                    throw new IcapTimeoutException(sprintf(
178                        'No connection was set up within %s seconds.',
179                        $setup,
180                    ));
181                }
182            }
183
184            throw $e;
185        } finally {
186            EventLoop::cancel($deadline);
187            if ($cancellation !== null && $watch !== null) {
188                $cancellation->unsubscribe($watch);
189            }
190        }
191
192        // The session gives each IO call (write / readResponse) the full
193        // streamTimeout window — no session-lifetime timer.
194        return new AmpTransportSession(
195            config: $config,
196            socket: $socket,
197            streamTimeout: $config->getStreamTimeout(),
198            userCancellation: $cancellation,
199            maxResponseSize: $config->getMaxResponseSize(),
200            maxHeaderLineLength: $config->getMaxHeaderLineLength(),
201            pool: $this->pool,
202        );
203    }
204
205    private function acquireSocket(Config $config, Cancellation $cancellation): SocketInterface
206    {
207        if ($this->pool !== null) {
208            return $this->pool->acquire($config, $cancellation);
209        }
210
211        return SocketOpener::open($config, $cancellation);
212    }
213}