Skip to content

Commit 766a1c6

Browse files
nicolas-grekasnicolas-grekas
committed
[HttpClient] add AsyncDecoratorTrait to ease processing responses without breaking async
1 parent bf53b26 commit 766a1c6

16 files changed

+979
-187
lines changed

‎src/Symfony/Component/HttpClient/AsyncDecoratorTrait.php

Lines changed: 54 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,54 @@
1+
<?php
2+
3+
/*
4+
* This file is part of the Symfony package.
5+
*
6+
* (c) Fabien Potencier <fabien@symfony.com>
7+
*
8+
* For the full copyright and license information, please view the LICENSE
9+
* file that was distributed with this source code.
10+
*/
11+
12+
namespace Symfony\Component\HttpClient;
13+
14+
use Symfony\Component\HttpClient\Response\AsyncResponse;
15+
use Symfony\Component\HttpClient\Response\ResponseStream;
16+
use Symfony\Contracts\HttpClient\HttpClientInterface;
17+
use Symfony\Contracts\HttpClient\ResponseInterface;
18+
use Symfony\Contracts\HttpClient\ResponseStreamInterface;
19+
20+
/**
21+
* Eases with processing responses while streaming them.
22+
*
23+
* @author Nicolas Grekas <p@tchwork.com>
24+
*/
25+
trait AsyncDecoratorTrait
26+
{
27+
private $client;
28+
29+
public function __construct(HttpClientInterface $client = null)
30+
{
31+
$this->client = $client ?? HttpClient::create();
32+
}
33+
34+
/**
35+
* {@inheritdoc}
36+
*
37+
* @return AsyncResponse
38+
*/
39+
abstract public function request(string $method, string $url, array $options = []): ResponseInterface;
40+
41+
/**
42+
* {@inheritdoc}
43+
*/
44+
public function stream($responses, float $timeout = null): ResponseStreamInterface
45+
{
46+
if ($responses instanceof AsyncResponse) {
47+
$responses = [$responses];
48+
} elseif (!is_iterable($responses)) {
49+
throw new \TypeError(sprintf('"%s()" expects parameter 1 to be an iterable of AsyncResponse objects, "%s" given.', __METHOD__, get_debug_type($responses)));
50+
}
51+
52+
return new ResponseStream(AsyncResponse::stream($responses, $timeout, static::class));
53+
}
54+
}

‎src/Symfony/Component/HttpClient/CHANGELOG.md

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,11 @@
11
CHANGELOG
22
=========
33

4+
5.2.0
5+
-----
6+
7+
* added `AsyncDecoratorTrait` to ease processing responses without breaking async
8+
49
5.1.0
510
-----
611

‎src/Symfony/Component/HttpClient/Chunk/ErrorChunk.php

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -111,8 +111,12 @@ public function getError(): ?string
111111
/**
112112
* @return bool Whether the wrapped error has been thrown or not
113113
*/
114-
public function didThrow(): bool
114+
public function didThrow(bool $didThrow = null): bool
115115
{
116+
if (null !== $didThrow && $this->didThrow !== $didThrow) {
117+
return !$this->didThrow = $didThrow;
118+
}
119+
116120
return $this->didThrow;
117121
}
118122

‎src/Symfony/Component/HttpClient/Internal/HttplugWaitLoop.php

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,7 @@
1515
use Psr\Http\Message\ResponseFactoryInterface;
1616
use Psr\Http\Message\ResponseInterface as Psr7ResponseInterface;
1717
use Psr\Http\Message\StreamFactoryInterface;
18-
use Symfony\Component\HttpClient\Response\ResponseTrait;
18+
use Symfony\Component\HttpClient\Response\CommonResponseTrait;
1919
use Symfony\Component\HttpClient\Response\StreamWrapper;
2020
use Symfony\Contracts\HttpClient\Exception\TransportExceptionInterface;
2121
use Symfony\Contracts\HttpClient\HttpClientInterface;
@@ -119,7 +119,7 @@ public function createPsr7Response(ResponseInterface $response, bool $buffer = f
119119
}
120120
}
121121

122-
if (isset(class_uses($response)[ResponseTrait::class])) {
122+
if (isset(class_uses($response)[CommonResponseTrait::class])) {
123123
$body = $this->streamFactory->createStreamFromResource($response->toStream(false));
124124
} elseif (!$buffer) {
125125
$body = $this->streamFactory->createStreamFromResource(StreamWrapper::createResource($response, $this->client));

‎src/Symfony/Component/HttpClient/Psr18Client.php

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,7 @@
2727
use Psr\Http\Message\StreamInterface;
2828
use Psr\Http\Message\UriFactoryInterface;
2929
use Psr\Http\Message\UriInterface;
30-
use Symfony\Component\HttpClient\Response\ResponseTrait;
30+
use Symfony\Component\HttpClient\Response\CommonResponseTrait;
3131
use Symfony\Component\HttpClient\Response\StreamWrapper;
3232
use Symfony\Contracts\HttpClient\Exception\TransportExceptionInterface;
3333
use Symfony\Contracts\HttpClient\HttpClientInterface;
@@ -104,7 +104,7 @@ public function sendRequest(RequestInterface $request): ResponseInterface
104104
}
105105
}
106106

107-
$body = isset(class_uses($response)[ResponseTrait::class]) ? $response->toStream(false) : StreamWrapper::createResource($response, $this->client);
107+
$body = isset(class_uses($response)[CommonResponseTrait::class]) ? $response->toStream(false) : StreamWrapper::createResource($response, $this->client);
108108
$body = $this->streamFactory->createStreamFromResource($body);
109109

110110
if ($body->isSeekable()) {

‎src/Symfony/Component/HttpClient/Response/AmpResponse.php

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -35,7 +35,8 @@
3535
*/
3636
final class AmpResponse implements ResponseInterface
3737
{
38-
use ResponseTrait;
38+
use CommonResponseTrait;
39+
use TransportResponseTrait;
3940

4041
private $multi;
4142
private $options;

‎src/Symfony/Component/HttpClient/Response/AsyncContext.php

Lines changed: 175 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,175 @@
1+
<?php
2+
3+
/*
4+
* This file is part of the Symfony package.
5+
*
6+
* (c) Fabien Potencier <fabien@symfony.com>
7+
*
8+
* For the full copyright and license information, please view the LICENSE
9+
* file that was distributed with this source code.
10+
*/
11+
12+
namespace Symfony\Component\HttpClient\Response;
13+
14+
use Symfony\Component\HttpClient\Chunk\DataChunk;
15+
use Symfony\Component\HttpClient\Chunk\LastChunk;
16+
use Symfony\Contracts\HttpClient\ChunkInterface;
17+
use Symfony\Contracts\HttpClient\HttpClientInterface;
18+
use Symfony\Contracts\HttpClient\ResponseInterface;
19+
20+
/**
21+
* A DTO to work with AsyncResponse.
22+
*
23+
* @author Nicolas Grekas <p@tchwork.com>
24+
*/
25+
final class AsyncContext
26+
{
27+
private $passthru;
28+
private $client;
29+
private $response;
30+
private $info = [];
31+
private $content;
32+
private $offset;
33+
34+
public function __construct(&$passthru, HttpClientInterface $client, ResponseInterface &$response, array &$info, $content, int $offset)
35+
{
36+
$this->passthru = &$passthru;
37+
$this->client = $client;
38+
$this->response = &$response;
39+
$this->info = &$info;
40+
$this->content = $content;
41+
$this->offset = $offset;
42+
}
43+
44+
/**
45+
* Returns the HTTP status without consuming the response.
46+
*/
47+
public function getStatusCode(): int
48+
{
49+
return $this->response->getInfo('http_code');
50+
}
51+
52+
/**
53+
* Returns the headers without consuming the response.
54+
*/
55+
public function getHeaders(): array
56+
{
57+
$headers = [];
58+
59+
foreach ($this->response->getInfo('response_headers') as $h) {
60+
if (11 <= \strlen($h) && '/' === $h[4] && preg_match('#^HTTP/\d+(?:\.\d+)? ([123456789]\d\d)(?: |$)#', $h, $m)) {
61+
$headers = [];
62+
} elseif (2 === \count($m = explode(':', $h, 2))) {
63+
$headers[strtolower($m[0])][] = ltrim($m[1]);
64+
}
65+
}
66+
67+
return $headers;
68+
}
69+
70+
/**
71+
* @return resource|null The PHP stream resource where the content is buffered, if it is
72+
*/
73+
public function getContent()
74+
{
75+
return $this->content;
76+
}
77+
78+
/**
79+
* Creates a new chunk of content.
80+
*/
81+
public function createChunk(string $data): ChunkInterface
82+
{
83+
return new DataChunk($this->offset, $data);
84+
}
85+
86+
/**
87+
* Pauses the request for the given number of seconds.
88+
*/
89+
public function pause(float $duration): void
90+
{
91+
if (\is_callable($pause = $this->response->getInfo('pause_handler'))) {
92+
$pause($duration);
93+
} elseif (0 < $duration) {
94+
usleep(1E6 * $duration);
95+
}
96+
}
97+
98+
/**
99+
* Cancels the request and returns the last chunk to yield.
100+
*/
101+
public function cancel(): ChunkInterface
102+
{
103+
$this->info['canceled'] = true;
104+
$this->info['error'] = 'Response has been canceled.';
105+
$this->response->cancel();
106+
107+
return new LastChunk();
108+
}
109+
110+
/**
111+
* Returns the current info of the response.
112+
*/
113+
public function getInfo(string $type = null)
114+
{
115+
if (null !== $type) {
116+
return $this->info[$type] ?? $this->response->getInfo($type);
117+
}
118+
119+
return $this->info + $this->response->getInfo();
120+
}
121+
122+
/**
123+
* Attaches an info to the response.
124+
*/
125+
public function setInfo(string $type, $value): self
126+
{
127+
if ('canceled' === $type && $value !== $this->info['canceled']) {
128+
throw new \LogicException('You cannot set the "canceled" info directly.');
129+
}
130+
131+
if (null === $value) {
132+
unset($this->info[$type]);
133+
} else {
134+
$this->info[$type] = $value;
135+
}
136+
137+
return $this;
138+
}
139+
140+
/**
141+
* Returns the currently processed response.
142+
*/
143+
public function getResponse(): ResponseInterface
144+
{
145+
return $this->response;
146+
}
147+
148+
/**
149+
* Replaces the currently processed response by doing a new request.
150+
*/
151+
public function replaceRequest(string $method, string $url, array $options = []): ResponseInterface
152+
{
153+
$this->info['previous_info'][] = $this->response->getInfo();
154+
155+
return $this->response = $this->client->request($method, $url, ['buffer' => false] + $options);
156+
}
157+
158+
/**
159+
* Replaces the currently processed response by another one.
160+
*/
161+
public function replaceResponse(ResponseInterface $response): ResponseInterface
162+
{
163+
$this->info['previous_info'][] = $this->response->getInfo();
164+
165+
return $this->response = $response;
166+
}
167+
168+
/**
169+
* Replaces or removes the chunk filter iterator.
170+
*/
171+
public function passthru(callable $passthru = null): void
172+
{
173+
$this->passthru = $passthru;
174+
}
175+
}

0 commit comments

Comments
 (0)