AmpResponse.php 17 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461
  1. <?php
  2. /*
  3. * This file is part of the Symfony package.
  4. *
  5. * (c) Fabien Potencier <fabien@symfony.com>
  6. *
  7. * For the full copyright and license information, please view the LICENSE
  8. * file that was distributed with this source code.
  9. */
  10. namespace Symfony\Component\HttpClient\Response;
  11. use Amp\ByteStream\StreamException;
  12. use Amp\CancellationTokenSource;
  13. use Amp\Coroutine;
  14. use Amp\Deferred;
  15. use Amp\Http\Client\HttpException;
  16. use Amp\Http\Client\Request;
  17. use Amp\Http\Client\Response;
  18. use Amp\Loop;
  19. use Amp\Promise;
  20. use Amp\Success;
  21. use Psr\Log\LoggerInterface;
  22. use Symfony\Component\HttpClient\Chunk\FirstChunk;
  23. use Symfony\Component\HttpClient\Chunk\InformationalChunk;
  24. use Symfony\Component\HttpClient\Exception\InvalidArgumentException;
  25. use Symfony\Component\HttpClient\Exception\TransportException;
  26. use Symfony\Component\HttpClient\HttpClientTrait;
  27. use Symfony\Component\HttpClient\Internal\AmpBody;
  28. use Symfony\Component\HttpClient\Internal\AmpClientState;
  29. use Symfony\Component\HttpClient\Internal\Canary;
  30. use Symfony\Component\HttpClient\Internal\ClientState;
  31. use Symfony\Contracts\HttpClient\ResponseInterface;
  32. /**
  33. * @author Nicolas Grekas <p@tchwork.com>
  34. *
  35. * @internal
  36. */
  37. final class AmpResponse implements ResponseInterface, StreamableInterface
  38. {
  39. use CommonResponseTrait;
  40. use TransportResponseTrait;
  41. private static string $nextId = 'a';
  42. private $multi;
  43. private ?array $options;
  44. private $canceller;
  45. private \Closure $onProgress;
  46. private static ?string $delay = null;
  47. /**
  48. * @internal
  49. */
  50. public function __construct(AmpClientState $multi, Request $request, array $options, ?LoggerInterface $logger)
  51. {
  52. $this->multi = $multi;
  53. $this->options = &$options;
  54. $this->logger = $logger;
  55. $this->timeout = $options['timeout'];
  56. $this->shouldBuffer = $options['buffer'];
  57. if ($this->inflate = \extension_loaded('zlib') && !$request->hasHeader('accept-encoding')) {
  58. $request->setHeader('Accept-Encoding', 'gzip');
  59. }
  60. $this->initializer = static function (self $response) {
  61. return null !== $response->options;
  62. };
  63. $info = &$this->info;
  64. $headers = &$this->headers;
  65. $canceller = $this->canceller = new CancellationTokenSource();
  66. $handle = &$this->handle;
  67. $info['url'] = (string) $request->getUri();
  68. $info['http_method'] = $request->getMethod();
  69. $info['start_time'] = null;
  70. $info['redirect_url'] = null;
  71. $info['redirect_time'] = 0.0;
  72. $info['redirect_count'] = 0;
  73. $info['size_upload'] = 0.0;
  74. $info['size_download'] = 0.0;
  75. $info['upload_content_length'] = -1.0;
  76. $info['download_content_length'] = -1.0;
  77. $info['user_data'] = $options['user_data'];
  78. $info['max_duration'] = $options['max_duration'];
  79. $info['debug'] = '';
  80. $onProgress = $options['on_progress'] ?? static function () {};
  81. $onProgress = $this->onProgress = static function () use (&$info, $onProgress) {
  82. $info['total_time'] = microtime(true) - $info['start_time'];
  83. $onProgress((int) $info['size_download'], ((int) (1 + $info['download_content_length']) ?: 1) - 1, (array) $info);
  84. };
  85. $pauseDeferred = new Deferred();
  86. $pause = new Success();
  87. $throttleWatcher = null;
  88. $this->id = $id = self::$nextId++;
  89. Loop::defer(static function () use ($request, $multi, &$id, &$info, &$headers, $canceller, &$options, $onProgress, &$handle, $logger, &$pause) {
  90. return new Coroutine(self::generateResponse($request, $multi, $id, $info, $headers, $canceller, $options, $onProgress, $handle, $logger, $pause));
  91. });
  92. $info['pause_handler'] = static function (float $duration) use (&$throttleWatcher, &$pauseDeferred, &$pause) {
  93. if (null !== $throttleWatcher) {
  94. Loop::cancel($throttleWatcher);
  95. }
  96. $pause = $pauseDeferred->promise();
  97. if ($duration <= 0) {
  98. $deferred = $pauseDeferred;
  99. $pauseDeferred = new Deferred();
  100. $deferred->resolve();
  101. } else {
  102. $throttleWatcher = Loop::delay(ceil(1000 * $duration), static function () use (&$pauseDeferred) {
  103. $deferred = $pauseDeferred;
  104. $pauseDeferred = new Deferred();
  105. $deferred->resolve();
  106. });
  107. }
  108. };
  109. $multi->lastTimeout = null;
  110. $multi->openHandles[$id] = $id;
  111. ++$multi->responseCount;
  112. $this->canary = new Canary(static function () use ($canceller, $multi, $id) {
  113. $canceller->cancel();
  114. unset($multi->openHandles[$id], $multi->handlesActivity[$id]);
  115. });
  116. }
  117. /**
  118. * {@inheritdoc}
  119. */
  120. public function getInfo(string $type = null): mixed
  121. {
  122. return null !== $type ? $this->info[$type] ?? null : $this->info;
  123. }
  124. public function __sleep(): array
  125. {
  126. throw new \BadMethodCallException('Cannot serialize '.__CLASS__);
  127. }
  128. public function __wakeup()
  129. {
  130. throw new \BadMethodCallException('Cannot unserialize '.__CLASS__);
  131. }
  132. public function __destruct()
  133. {
  134. try {
  135. $this->doDestruct();
  136. } finally {
  137. // Clear the DNS cache when all requests completed
  138. if (0 >= --$this->multi->responseCount) {
  139. $this->multi->responseCount = 0;
  140. $this->multi->dnsCache = [];
  141. }
  142. }
  143. }
  144. /**
  145. * {@inheritdoc}
  146. */
  147. private static function schedule(self $response, array &$runningResponses): void
  148. {
  149. if (isset($runningResponses[0])) {
  150. $runningResponses[0][1][$response->id] = $response;
  151. } else {
  152. $runningResponses[0] = [$response->multi, [$response->id => $response]];
  153. }
  154. if (!isset($response->multi->openHandles[$response->id])) {
  155. $response->multi->handlesActivity[$response->id][] = null;
  156. $response->multi->handlesActivity[$response->id][] = null !== $response->info['error'] ? new TransportException($response->info['error']) : null;
  157. }
  158. }
  159. /**
  160. * {@inheritdoc}
  161. *
  162. * @param AmpClientState $multi
  163. */
  164. private static function perform(ClientState $multi, array &$responses = null): void
  165. {
  166. if ($responses) {
  167. foreach ($responses as $response) {
  168. try {
  169. if ($response->info['start_time']) {
  170. $response->info['total_time'] = microtime(true) - $response->info['start_time'];
  171. ($response->onProgress)();
  172. }
  173. } catch (\Throwable $e) {
  174. $multi->handlesActivity[$response->id][] = null;
  175. $multi->handlesActivity[$response->id][] = $e;
  176. }
  177. }
  178. }
  179. }
  180. /**
  181. * {@inheritdoc}
  182. *
  183. * @param AmpClientState $multi
  184. */
  185. private static function select(ClientState $multi, float $timeout): int
  186. {
  187. $timeout += microtime(true);
  188. self::$delay = Loop::defer(static function () use ($timeout) {
  189. if (0 < $timeout -= microtime(true)) {
  190. self::$delay = Loop::delay(ceil(1000 * $timeout), [Loop::class, 'stop']);
  191. } else {
  192. Loop::stop();
  193. }
  194. });
  195. Loop::run();
  196. return null === self::$delay ? 1 : 0;
  197. }
  198. private static function generateResponse(Request $request, AmpClientState $multi, string $id, array &$info, array &$headers, CancellationTokenSource $canceller, array &$options, \Closure $onProgress, &$handle, ?LoggerInterface $logger, Promise &$pause)
  199. {
  200. $request->setInformationalResponseHandler(static function (Response $response) use ($multi, $id, &$info, &$headers) {
  201. self::addResponseHeaders($response, $info, $headers);
  202. $multi->handlesActivity[$id][] = new InformationalChunk($response->getStatus(), $response->getHeaders());
  203. self::stopLoop();
  204. });
  205. try {
  206. /* @var Response $response */
  207. if (null === $response = yield from self::getPushedResponse($request, $multi, $info, $headers, $options, $logger)) {
  208. $logger && $logger->info(sprintf('Request: "%s %s"', $info['http_method'], $info['url']));
  209. $response = yield from self::followRedirects($request, $multi, $info, $headers, $canceller, $options, $onProgress, $handle, $logger, $pause);
  210. }
  211. $options = null;
  212. $multi->handlesActivity[$id][] = new FirstChunk();
  213. if ('HEAD' === $response->getRequest()->getMethod() || \in_array($info['http_code'], [204, 304], true)) {
  214. $multi->handlesActivity[$id][] = null;
  215. $multi->handlesActivity[$id][] = null;
  216. self::stopLoop();
  217. return;
  218. }
  219. if ($response->hasHeader('content-length')) {
  220. $info['download_content_length'] = (float) $response->getHeader('content-length');
  221. }
  222. $body = $response->getBody();
  223. while (true) {
  224. self::stopLoop();
  225. yield $pause;
  226. if (null === $data = yield $body->read()) {
  227. break;
  228. }
  229. $info['size_download'] += \strlen($data);
  230. $multi->handlesActivity[$id][] = $data;
  231. }
  232. $multi->handlesActivity[$id][] = null;
  233. $multi->handlesActivity[$id][] = null;
  234. } catch (\Throwable $e) {
  235. $multi->handlesActivity[$id][] = null;
  236. $multi->handlesActivity[$id][] = $e;
  237. } finally {
  238. $info['download_content_length'] = $info['size_download'];
  239. }
  240. self::stopLoop();
  241. }
  242. private static function followRedirects(Request $originRequest, AmpClientState $multi, array &$info, array &$headers, CancellationTokenSource $canceller, array $options, \Closure $onProgress, &$handle, ?LoggerInterface $logger, Promise &$pause)
  243. {
  244. yield $pause;
  245. $originRequest->setBody(new AmpBody($options['body'], $info, $onProgress));
  246. $response = yield $multi->request($options, $originRequest, $canceller->getToken(), $info, $onProgress, $handle);
  247. $previousUrl = null;
  248. while (true) {
  249. self::addResponseHeaders($response, $info, $headers);
  250. $status = $response->getStatus();
  251. if (!\in_array($status, [301, 302, 303, 307, 308], true) || null === $location = $response->getHeader('location')) {
  252. return $response;
  253. }
  254. $urlResolver = new class() {
  255. use HttpClientTrait {
  256. parseUrl as public;
  257. resolveUrl as public;
  258. }
  259. };
  260. try {
  261. $previousUrl = $previousUrl ?? $urlResolver::parseUrl($info['url']);
  262. $location = $urlResolver::parseUrl($location);
  263. $location = $urlResolver::resolveUrl($location, $previousUrl);
  264. $info['redirect_url'] = implode('', $location);
  265. } catch (InvalidArgumentException $e) {
  266. return $response;
  267. }
  268. if (0 >= $options['max_redirects'] || $info['redirect_count'] >= $options['max_redirects']) {
  269. return $response;
  270. }
  271. $logger && $logger->info(sprintf('Redirecting: "%s %s"', $status, $info['url']));
  272. try {
  273. // Discard body of redirects
  274. while (null !== yield $response->getBody()->read()) {
  275. }
  276. } catch (HttpException|StreamException $e) {
  277. // Ignore streaming errors on previous responses
  278. }
  279. ++$info['redirect_count'];
  280. $info['url'] = $info['redirect_url'];
  281. $info['redirect_url'] = null;
  282. $previousUrl = $location;
  283. $request = new Request($info['url'], $info['http_method']);
  284. $request->setProtocolVersions($originRequest->getProtocolVersions());
  285. $request->setTcpConnectTimeout($originRequest->getTcpConnectTimeout());
  286. $request->setTlsHandshakeTimeout($originRequest->getTlsHandshakeTimeout());
  287. $request->setTransferTimeout($originRequest->getTransferTimeout());
  288. if (\in_array($status, [301, 302, 303], true)) {
  289. $originRequest->removeHeader('transfer-encoding');
  290. $originRequest->removeHeader('content-length');
  291. $originRequest->removeHeader('content-type');
  292. // Do like curl and browsers: turn POST to GET on 301, 302 and 303
  293. if ('POST' === $response->getRequest()->getMethod() || 303 === $status) {
  294. $info['http_method'] = 'HEAD' === $response->getRequest()->getMethod() ? 'HEAD' : 'GET';
  295. $request->setMethod($info['http_method']);
  296. }
  297. } else {
  298. $request->setBody(AmpBody::rewind($response->getRequest()->getBody()));
  299. }
  300. foreach ($originRequest->getRawHeaders() as [$name, $value]) {
  301. $request->setHeader($name, $value);
  302. }
  303. if ($request->getUri()->getAuthority() !== $originRequest->getUri()->getAuthority()) {
  304. $request->removeHeader('authorization');
  305. $request->removeHeader('cookie');
  306. $request->removeHeader('host');
  307. }
  308. yield $pause;
  309. $response = yield $multi->request($options, $request, $canceller->getToken(), $info, $onProgress, $handle);
  310. $info['redirect_time'] = microtime(true) - $info['start_time'];
  311. }
  312. }
  313. private static function addResponseHeaders(Response $response, array &$info, array &$headers): void
  314. {
  315. $info['http_code'] = $response->getStatus();
  316. if ($headers) {
  317. $info['debug'] .= "< \r\n";
  318. $headers = [];
  319. }
  320. $h = sprintf('HTTP/%s %s %s', $response->getProtocolVersion(), $response->getStatus(), $response->getReason());
  321. $info['debug'] .= "< {$h}\r\n";
  322. $info['response_headers'][] = $h;
  323. foreach ($response->getRawHeaders() as [$name, $value]) {
  324. $headers[strtolower($name)][] = $value;
  325. $h = $name.': '.$value;
  326. $info['debug'] .= "< {$h}\r\n";
  327. $info['response_headers'][] = $h;
  328. }
  329. $info['debug'] .= "< \r\n";
  330. }
  331. /**
  332. * Accepts pushed responses only if their headers related to authentication match the request.
  333. */
  334. private static function getPushedResponse(Request $request, AmpClientState $multi, array &$info, array &$headers, array $options, ?LoggerInterface $logger)
  335. {
  336. if ('' !== $options['body']) {
  337. return null;
  338. }
  339. $authority = $request->getUri()->getAuthority();
  340. foreach ($multi->pushedResponses[$authority] ?? [] as $i => [$pushedUrl, $pushDeferred, $pushedRequest, $pushedResponse, $parentOptions]) {
  341. if ($info['url'] !== $pushedUrl || $info['http_method'] !== $pushedRequest->getMethod()) {
  342. continue;
  343. }
  344. foreach ($parentOptions as $k => $v) {
  345. if ($options[$k] !== $v) {
  346. continue 2;
  347. }
  348. }
  349. foreach (['authorization', 'cookie', 'range', 'proxy-authorization'] as $k) {
  350. if ($pushedRequest->getHeaderArray($k) !== $request->getHeaderArray($k)) {
  351. continue 2;
  352. }
  353. }
  354. $response = yield $pushedResponse;
  355. foreach ($response->getHeaderArray('vary') as $vary) {
  356. foreach (preg_split('/\s*+,\s*+/', $vary) as $v) {
  357. if ('*' === $v || ($pushedRequest->getHeaderArray($v) !== $request->getHeaderArray($v) && 'accept-encoding' !== strtolower($v))) {
  358. $logger && $logger->debug(sprintf('Skipping pushed response: "%s"', $info['url']));
  359. continue 3;
  360. }
  361. }
  362. }
  363. $pushDeferred->resolve();
  364. $logger && $logger->debug(sprintf('Accepting pushed response: "%s %s"', $info['http_method'], $info['url']));
  365. self::addResponseHeaders($response, $info, $headers);
  366. unset($multi->pushedResponses[$authority][$i]);
  367. if (!$multi->pushedResponses[$authority]) {
  368. unset($multi->pushedResponses[$authority]);
  369. }
  370. return $response;
  371. }
  372. }
  373. private static function stopLoop(): void
  374. {
  375. if (null !== self::$delay) {
  376. Loop::cancel(self::$delay);
  377. self::$delay = null;
  378. }
  379. Loop::defer([Loop::class, 'stop']);
  380. }
  381. }