NativeResponse.php 14 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401
  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 Psr\Log\LoggerInterface;
  12. use Symfony\Component\HttpClient\Chunk\FirstChunk;
  13. use Symfony\Component\HttpClient\Exception\TransportException;
  14. use Symfony\Component\HttpClient\Internal\Canary;
  15. use Symfony\Component\HttpClient\Internal\ClientState;
  16. use Symfony\Component\HttpClient\Internal\NativeClientState;
  17. use Symfony\Contracts\HttpClient\ResponseInterface;
  18. /**
  19. * @author Nicolas Grekas <p@tchwork.com>
  20. *
  21. * @internal
  22. */
  23. final class NativeResponse implements ResponseInterface, StreamableInterface
  24. {
  25. use CommonResponseTrait;
  26. use TransportResponseTrait;
  27. /**
  28. * @var resource
  29. */
  30. private $context;
  31. /**
  32. * @var string
  33. */
  34. private $url;
  35. private $resolver;
  36. private $onProgress;
  37. /**
  38. * @var int|null
  39. */
  40. private $remaining;
  41. /**
  42. * @var resource|null
  43. */
  44. private $buffer;
  45. /**
  46. * @var \Symfony\Component\HttpClient\Internal\NativeClientState
  47. */
  48. private $multi;
  49. /**
  50. * @var float
  51. */
  52. private $pauseExpiry = 0.0;
  53. /**
  54. * @internal
  55. * @param \Symfony\Component\HttpClient\Internal\NativeClientState $multi
  56. * @param string $url
  57. * @param mixed[] $options
  58. * @param mixed[] $info
  59. * @param callable $resolver
  60. * @param callable|null $onProgress
  61. * @param \Psr\Log\LoggerInterface|null $logger
  62. */
  63. public function __construct($multi, $context, $url, $options, &$info, $resolver, $onProgress, $logger)
  64. {
  65. $this->multi = $multi;
  66. $this->id = $id = (int) $context;
  67. $this->context = $context;
  68. $this->url = $url;
  69. $this->logger = $logger;
  70. $this->timeout = $options['timeout'];
  71. $this->info = &$info;
  72. $this->resolver = $resolver;
  73. $this->onProgress = $onProgress;
  74. $this->inflate = !isset($options['normalized_headers']['accept-encoding']);
  75. $this->shouldBuffer = $options['buffer'] ?? true;
  76. // Temporary resource to dechunk the response stream
  77. $this->buffer = fopen('php://temp', 'w+');
  78. $info['original_url'] = implode('', $info['url']);
  79. $info['user_data'] = $options['user_data'];
  80. $info['max_duration'] = $options['max_duration'];
  81. ++$multi->responseCount;
  82. $this->initializer = static function (self $response) {
  83. return null === $response->remaining;
  84. };
  85. $pauseExpiry = &$this->pauseExpiry;
  86. $info['pause_handler'] = static function (float $duration) use (&$pauseExpiry) {
  87. $pauseExpiry = 0 < $duration ? microtime(true) + $duration : 0;
  88. };
  89. $this->canary = new Canary(static function () use ($multi, $id) {
  90. if (null !== ($host = $multi->openHandles[$id][6] ?? null) && 0 >= --$multi->hosts[$host]) {
  91. unset($multi->hosts[$host]);
  92. }
  93. unset($multi->openHandles[$id], $multi->handlesActivity[$id]);
  94. });
  95. }
  96. /**
  97. * @return mixed
  98. * @param string|null $type
  99. */
  100. public function getInfo($type = null)
  101. {
  102. if (!$info = $this->finalInfo) {
  103. $info = $this->info;
  104. $info['url'] = implode('', $info['url']);
  105. unset($info['size_body'], $info['request_header']);
  106. if (null === $this->buffer) {
  107. $this->finalInfo = $info;
  108. }
  109. }
  110. return null !== $type ? $info[$type] ?? null : $info;
  111. }
  112. public function __destruct()
  113. {
  114. try {
  115. $this->doDestruct();
  116. } finally {
  117. // Clear the DNS cache when all requests completed
  118. if (0 >= --$this->multi->responseCount) {
  119. $this->multi->responseCount = 0;
  120. $this->multi->dnsCache = [];
  121. }
  122. }
  123. }
  124. private function open()
  125. {
  126. $url = $this->url;
  127. set_error_handler(function ($type, $msg) use (&$url) {
  128. if (\E_NOTICE !== $type || 'fopen(): Content-type not specified assuming application/x-www-form-urlencoded' !== $msg) {
  129. throw new TransportException($msg);
  130. }
  131. ($logger = $this->logger) ? $logger->info(sprintf('%s for "%s".', $msg, $url ?? $this->url)) : null;
  132. });
  133. try {
  134. $this->info['start_time'] = microtime(true);
  135. [$resolver, $url] = ($this->resolver)($this->multi);
  136. while (true) {
  137. $context = stream_context_get_options($this->context);
  138. if ($proxy = $context['http']['proxy'] ?? null) {
  139. $this->info['debug'] .= "* Establish HTTP proxy tunnel to {$proxy}\n";
  140. $this->info['request_header'] = $url;
  141. } else {
  142. $this->info['debug'] .= "* Trying {$this->info['primary_ip']}...\n";
  143. $this->info['request_header'] = $this->info['url']['path'].$this->info['url']['query'];
  144. }
  145. $this->info['request_header'] = sprintf("> %s %s HTTP/%s \r\n", $context['http']['method'], $this->info['request_header'], $context['http']['protocol_version']);
  146. $this->info['request_header'] .= implode("\r\n", $context['http']['header'])."\r\n\r\n";
  147. if (\array_key_exists('peer_name', $context['ssl']) && null === $context['ssl']['peer_name']) {
  148. unset($context['ssl']['peer_name']);
  149. $this->context = stream_context_create([], ['options' => $context] + stream_context_get_params($this->context));
  150. }
  151. // Send request and follow redirects when needed
  152. $this->handle = $h = fopen($url, 'r', false, $this->context);
  153. self::addResponseHeaders(stream_get_meta_data($h)['wrapper_data'], $this->info, $this->headers, $this->info['debug']);
  154. $url = $resolver($this->multi, $this->headers['location'][0] ?? null, $this->context);
  155. if (null === $url) {
  156. break;
  157. }
  158. ($logger = $this->logger) ? $logger->info(sprintf('Redirecting: "%s %s"', $this->info['http_code'], $url ?? $this->url)) : null;
  159. }
  160. } catch (\Throwable $e) {
  161. $this->close();
  162. $this->multi->handlesActivity[$this->id][] = null;
  163. $this->multi->handlesActivity[$this->id][] = $e;
  164. return;
  165. } finally {
  166. $this->info['pretransfer_time'] = $this->info['total_time'] = microtime(true) - $this->info['start_time'];
  167. restore_error_handler();
  168. }
  169. if (isset($context['ssl']['capture_peer_cert_chain']) && isset(($context = stream_context_get_options($this->context))['ssl']['peer_certificate_chain'])) {
  170. $this->info['peer_certificate_chain'] = $context['ssl']['peer_certificate_chain'];
  171. }
  172. stream_set_blocking($h, false);
  173. $this->context = $this->resolver = null;
  174. // Create dechunk buffers
  175. if (isset($this->headers['content-length'])) {
  176. $this->remaining = (int) $this->headers['content-length'][0];
  177. } elseif ('chunked' === ($this->headers['transfer-encoding'][0] ?? null)) {
  178. stream_filter_append($this->buffer, 'dechunk', \STREAM_FILTER_WRITE);
  179. $this->remaining = -1;
  180. } else {
  181. $this->remaining = -2;
  182. }
  183. $this->multi->handlesActivity[$this->id] = [new FirstChunk()];
  184. if ('HEAD' === $context['http']['method'] || \in_array($this->info['http_code'], [204, 304], true)) {
  185. $this->multi->handlesActivity[$this->id][] = null;
  186. $this->multi->handlesActivity[$this->id][] = null;
  187. return;
  188. }
  189. $host = parse_url($this->info['redirect_url'] ?? $this->url, \PHP_URL_HOST);
  190. $this->multi->lastTimeout = null;
  191. $this->multi->openHandles[$this->id] = [&$this->pauseExpiry, $h, $this->buffer, $this->onProgress, &$this->remaining, &$this->info, $host];
  192. $this->multi->hosts[$host] = 1 + ($this->multi->hosts[$host] ?? 0);
  193. }
  194. private function close()
  195. {
  196. $this->canary->cancel();
  197. $this->handle = $this->buffer = $this->inflate = $this->onProgress = null;
  198. }
  199. /**
  200. * @param $this $response
  201. * @param mixed[] $runningResponses
  202. */
  203. private static function schedule($response, &$runningResponses)
  204. {
  205. if (!isset($runningResponses[$i = $response->multi->id])) {
  206. $runningResponses[$i] = [$response->multi, []];
  207. }
  208. $runningResponses[$i][1][$response->id] = $response;
  209. if (null === $response->buffer) {
  210. // Response already completed
  211. $response->multi->handlesActivity[$response->id][] = null;
  212. $response->multi->handlesActivity[$response->id][] = null !== $response->info['error'] ? new TransportException($response->info['error']) : null;
  213. }
  214. }
  215. /**
  216. * @param \Symfony\Component\HttpClient\Internal\ClientState $multi
  217. * @param mixed[]|null $responses
  218. */
  219. private static function perform($multi, &$responses = null)
  220. {
  221. foreach ($multi->openHandles as $i => [$pauseExpiry, $h, $buffer, $onProgress]) {
  222. if ($pauseExpiry) {
  223. if (microtime(true) < $pauseExpiry) {
  224. continue;
  225. }
  226. $multi->openHandles[$i][0] = 0;
  227. }
  228. $hasActivity = false;
  229. $remaining = &$multi->openHandles[$i][4];
  230. $info = &$multi->openHandles[$i][5];
  231. $e = null;
  232. // Read incoming buffer and write it to the dechunk one
  233. try {
  234. if ($remaining && '' !== $data = (string) fread($h, 0 > $remaining ? 16372 : $remaining)) {
  235. fwrite($buffer, $data);
  236. $hasActivity = true;
  237. $multi->sleep = false;
  238. if (-1 !== $remaining) {
  239. $remaining -= \strlen($data);
  240. }
  241. }
  242. } catch (\Throwable $e) {
  243. $hasActivity = $onProgress = false;
  244. }
  245. if (!$hasActivity) {
  246. if ($onProgress) {
  247. try {
  248. // Notify the progress callback so that it can e.g. cancel
  249. // the request if the stream is inactive for too long
  250. $info['total_time'] = microtime(true) - $info['start_time'];
  251. $onProgress();
  252. } catch (\Throwable $e) {
  253. // no-op
  254. }
  255. }
  256. } elseif ('' !== $data = stream_get_contents($buffer, -1, 0)) {
  257. rewind($buffer);
  258. ftruncate($buffer, 0);
  259. if (null === $e) {
  260. $multi->handlesActivity[$i][] = $data;
  261. }
  262. }
  263. if (null !== $e || !$remaining || feof($h)) {
  264. // Stream completed
  265. $info['total_time'] = microtime(true) - $info['start_time'];
  266. $info['starttransfer_time'] = $info['starttransfer_time'] ?: $info['total_time'];
  267. if ($onProgress) {
  268. try {
  269. $onProgress(-1);
  270. } catch (\Throwable $e) {
  271. // no-op
  272. }
  273. }
  274. if (null === $e) {
  275. if (0 < $remaining) {
  276. $e = new TransportException(sprintf('Transfer closed with %s bytes remaining to read.', $remaining));
  277. } elseif (-1 === $remaining && fwrite($buffer, '-') && '' !== stream_get_contents($buffer, -1, 0)) {
  278. $e = new TransportException('Transfer closed with outstanding data remaining from chunked response.');
  279. }
  280. }
  281. $multi->handlesActivity[$i][] = null;
  282. $multi->handlesActivity[$i][] = $e;
  283. if (null !== ($host = $multi->openHandles[$i][6] ?? null) && 0 >= --$multi->hosts[$host]) {
  284. unset($multi->hosts[$host]);
  285. }
  286. unset($multi->openHandles[$i]);
  287. $multi->sleep = false;
  288. }
  289. }
  290. if (null === $responses) {
  291. return;
  292. }
  293. $maxHosts = $multi->maxHostConnections;
  294. foreach ($responses as $i => $response) {
  295. if (null !== $response->remaining || null === $response->buffer) {
  296. continue;
  297. }
  298. if ($response->pauseExpiry && microtime(true) < $response->pauseExpiry) {
  299. // Create empty open handles to tell we still have pending requests
  300. $multi->openHandles[$i] = [\INF, null, null, null];
  301. } elseif ($maxHosts && $maxHosts > ($multi->hosts[parse_url($response->url, \PHP_URL_HOST)] ?? 0)) {
  302. // Open the next pending request - this is a blocking operation so we do only one of them
  303. $response->open();
  304. $multi->sleep = false;
  305. self::perform($multi);
  306. $maxHosts = 0;
  307. }
  308. }
  309. }
  310. /**
  311. * @param \Symfony\Component\HttpClient\Internal\ClientState $multi
  312. * @param float $timeout
  313. */
  314. private static function select($multi, $timeout)
  315. {
  316. if (!$multi->sleep = !$multi->sleep) {
  317. return -1;
  318. }
  319. $_ = $handles = [];
  320. $now = null;
  321. foreach ($multi->openHandles as [$pauseExpiry, $h]) {
  322. if (null === $h) {
  323. continue;
  324. }
  325. if ($pauseExpiry && ($now = $now ?? microtime(true)) < $pauseExpiry) {
  326. $timeout = min($timeout, $pauseExpiry - $now);
  327. continue;
  328. }
  329. $handles[] = $h;
  330. }
  331. if (!$handles) {
  332. usleep((int) (1E6 * $timeout));
  333. return 0;
  334. }
  335. return stream_select($handles, $_, $_, (int) $timeout, (int) (1E6 * ($timeout - (int) $timeout)));
  336. }
  337. }