Code Coverage
 
Lines
Functions and Methods
Classes and Traits
Total
73.33%
22 / 30
42.86%
3 / 7
 CRAP
0.00%
0 / 1
AmpTransportSession
73.33%
22 / 30
42.86%
3 / 7
 17.72
0.00%
0 / 1
 __construct
100.00%
1 / 1
100.00%
1 / 1
 1
 write
100.00%
4 / 4
100.00%
1 / 1
 3
 readResponse
100.00%
8 / 8
100.00%
1 / 1
 1
 makeIoCancellation
75.00%
3 / 4
0.00%
0 / 1
 2.06
 release
71.43%
5 / 7
0.00%
0 / 1
 3.21
 close
0.00%
0 / 4
0.00%
0 / 1
 6
 assertActive
50.00%
1 / 2
0.00%
0 / 1
 2.50
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\CompositeCancellation;
25use Amp\Socket\Socket as SocketInterface;
26use Amp\TimeoutCancellation;
27use Ndrstmr\Icap\Config;
28
29/**
30 * One-acquired-socket binding used for multi-round-trip flows like
31 * the strict RFC 3507 §4.5 preview-continue exchange.
32 *
33 * Lifecycle:
34 *   - {@see AsyncAmpTransport::openSession()} acquires a socket from
35 *     the pool (or opens a fresh one).
36 *   - The caller writes / reads as many times as the protocol
37 *     requires.
38 *   - The caller calls {@see release()} on success (returns to the
39 *     pool when one is configured) or {@see close()} on failure.
40 */
41final class AmpTransportSession implements TransportSession
42{
43    private bool $disposed = false;
44
45    public function __construct(
46        private readonly Config $config,
47        private readonly SocketInterface $socket,
48        private readonly float $streamTimeout,
49        private readonly ?Cancellation $userCancellation,
50        private readonly int $maxResponseSize,
51        private readonly int $maxHeaderLineLength,
52        private readonly ?ConnectionPoolInterface $pool,
53    ) {
54    }
55
56    #[\Override]
57    public function write(iterable $chunks): void
58    {
59        $this->assertActive();
60        foreach ($chunks as $chunk) {
61            if ($chunk !== '') {
62                $this->socket->write($chunk);
63            }
64        }
65    }
66
67    #[\Override]
68    public function readResponse(): string
69    {
70        $this->assertActive();
71        $reader = new ResponseFrameReader(
72            maxResponseSize: $this->maxResponseSize,
73            maxHeaderLineLength: $this->maxHeaderLineLength,
74        );
75        $cancellation = $this->makeIoCancellation();
76        $socket = $this->socket;
77        return $reader->readFrom(static fn (): ?string => $socket->read($cancellation));
78    }
79
80    /**
81     * Creates a fresh TimeoutCancellation for a single IO phase,
82     * combined with the optional user-supplied Cancellation.
83     *
84     * Each call gets the full streamTimeout window — no session-lifetime
85     * timer that accumulates across multiple preview-continue round trips.
86     */
87    private function makeIoCancellation(): Cancellation
88    {
89        $timeout = new TimeoutCancellation($this->streamTimeout);
90        if ($this->userCancellation === null) {
91            return $timeout;
92        }
93        return new CompositeCancellation($this->userCancellation, $timeout);
94    }
95
96    #[\Override]
97    public function release(): void
98    {
99        if ($this->disposed) {
100            return;
101        }
102        $this->disposed = true;
103
104        if ($this->pool !== null) {
105            $this->pool->release($this->config, $this->socket);
106            return;
107        }
108        $this->socket->close();
109    }
110
111    #[\Override]
112    public function close(): void
113    {
114        if ($this->disposed) {
115            return;
116        }
117        $this->disposed = true;
118        $this->socket->close();
119    }
120
121    private function assertActive(): void
122    {
123        if ($this->disposed) {
124            throw new \LogicException('TransportSession has already been released or closed.');
125        }
126    }
127}