Code Coverage
 
Lines
Functions and Methods
Classes and Traits
Total
85.71%
6 / 7
75.00%
3 / 4
 CRAP
0.00%
0 / 1
SynchronousStreamTransport
85.71%
6 / 7
75.00%
3 / 4
 4.05
0.00%
0 / 1
 request
100.00%
1 / 1
100.00%
1 / 1
 1
 requestDiscardingBody
0.00%
0 / 1
0.00%
0 / 1
 2
 openSession
100.00%
1 / 1
100.00%
1 / 1
 1
 exchange
100.00%
4 / 4
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;
26
27/**
28 * Blocking transport using plain PHP stream sockets, for CLI and test usage. TLS is not supported; production
29 * deployments use {@see AsyncAmpTransport} (TLS, pooling, cancellation that interrupts a wait).
30 *
31 * Every exchange runs on a {@see SynchronousStreamSession}, which holds the limits, the cancellation and the framing.
32 * A request is a session of its own that is closed after its response, and its write keeps the limit of this
33 * transport: `streamTimeout` is the time the peer may go without taking a byte. {@see openSession()} hands the session
34 * out for an exchange of more than one round trip, such as a preview that the server answers with `100 Continue`
35 * (RFC 3507 ยง4.5); each of its writes has one window for the whole call ({@see TransportSession::write()}).
36 */
37final class SynchronousStreamTransport implements SessionAwareTransport, DiscardingTransport
38{
39    /**
40     * @param iterable<string> $rawRequest
41     * @return \Amp\Future<string>
42     */
43    #[\Override]
44    public function request(Config $config, iterable $rawRequest, ?Cancellation $cancellation = null): \Amp\Future
45    {
46        return $this->exchange($config, $rawRequest, $cancellation, static fn (SynchronousStreamSession $session): string => $session->readResponse());
47    }
48
49    /**
50     * @param iterable<string> $rawRequest
51     * @return \Amp\Future<IcapResponse>
52     */
53    #[\Override]
54    public function requestDiscardingBody(Config $config, iterable $rawRequest, int $maxResponseBytes, ?Cancellation $cancellation = null): \Amp\Future
55    {
56        return $this->exchange($config, $rawRequest, $cancellation, static fn (SynchronousStreamSession $session): IcapResponse => $session->readResponseDiscardingBody($maxResponseBytes));
57    }
58
59    #[\Override]
60    public function openSession(Config $config, ?Cancellation $cancellation = null): SynchronousStreamSession
61    {
62        return SynchronousStreamSession::open($config, $cancellation);
63    }
64
65    /**
66     * One request on a session of its own, and what the caller reads from it.
67     *
68     * @template T
69     *
70     * @param iterable<string>                               $rawRequest
71     * @param \Closure(SynchronousStreamSession): T          $read
72     *
73     * @return \Amp\Future<T>
74     */
75    private function exchange(Config $config, iterable $rawRequest, ?Cancellation $cancellation, \Closure $read): \Amp\Future
76    {
77        $session = $this->openSession($config, $cancellation);
78        try {
79            $session->sendRequest($rawRequest);
80
81            return \Amp\Future::complete($read($session));
82        } finally {
83            $session->close();
84        }
85    }
86}