SdkStreamHandler.php 13 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390
  1. <?php
  2. /**
  3. * Copyright (c) 2011-2018 Michael Dowling, https://github.com/mtdowling <mtdowling@gmail.com>
  4. * Permission is hereby granted, free of charge, to any person obtaining a copy
  5. * of this software and associated documentation files (the "Software"), to deal
  6. * in the Software without restriction, including without limitation the rights
  7. * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
  8. * copies of the Software, and to permit persons to whom the Software is
  9. * furnished to do so, subject to the following conditions:
  10. * The above copyright notice and this permission notice shall be included in
  11. * all copies or substantial portions of the Software.
  12. * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  13. * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  14. * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  15. * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  16. * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  17. * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
  18. * THE SOFTWARE.
  19. */
  20. namespace Obs\Internal\Common;
  21. use GuzzleHttp\Psr7;
  22. use GuzzleHttp\Exception\ConnectException;
  23. use GuzzleHttp\Exception\RequestException;
  24. use GuzzleHttp\Promise\FulfilledPromise;
  25. use GuzzleHttp\TransferStats;
  26. use Psr\Http\Message\RequestInterface;
  27. use Psr\Http\Message\ResponseInterface;
  28. use Psr\Http\Message\StreamInterface;
  29. class SdkStreamHandler
  30. {
  31. private $lastHeaders = [];
  32. public function __invoke(RequestInterface $request, array $options)
  33. {
  34. if (isset($options['delay'])) {
  35. usleep($options['delay'] * 1000);
  36. }
  37. $startTime = isset($options['on_stats']) ? microtime(true) : null;
  38. try {
  39. $request = $request->withoutHeader('Expect');
  40. if (0 === $request->getBody()->getSize()) {
  41. $request = $request->withHeader('Content-Length', 0);
  42. }
  43. return $this->createResponse(
  44. $request,
  45. $options,
  46. $this->createStream($request, $options),
  47. $startTime
  48. );
  49. } catch (\InvalidArgumentException $e) {
  50. throw $e;
  51. } catch (\Exception $e) {
  52. $message = $e->getMessage();
  53. if (strpos($message, 'getaddrinfo')
  54. || strpos($message, 'Connection refused')
  55. || strpos($message, "couldn't connect to host")
  56. ) {
  57. $e = new ConnectException($e->getMessage(), $request, $e);
  58. }
  59. $e = RequestException::wrapException($request, $e);
  60. $this->invokeStats($options, $request, $startTime, null, $e);
  61. return \GuzzleHttp\Promise\rejection_for($e);
  62. }
  63. }
  64. private function invokeStats(
  65. array $options,
  66. RequestInterface $request,
  67. $startTime,
  68. ResponseInterface $response = null,
  69. $error = null
  70. ) {
  71. if (isset($options['on_stats'])) {
  72. $stats = new TransferStats(
  73. $request,
  74. $response,
  75. microtime(true) - $startTime,
  76. $error,
  77. []
  78. );
  79. call_user_func($options['on_stats'], $stats);
  80. }
  81. }
  82. private function createResponse(
  83. RequestInterface $request,
  84. array $options,
  85. $stream,
  86. $startTime
  87. ) {
  88. $hdrs = $this->lastHeaders;
  89. $this->lastHeaders = [];
  90. $parts = explode(' ', array_shift($hdrs), 3);
  91. $ver = explode('/', $parts[0])[1];
  92. $status = $parts[1];
  93. $reason = isset($parts[2]) ? $parts[2] : null;
  94. $headers = \GuzzleHttp\headers_from_lines($hdrs);
  95. list($stream, $headers) = $this->checkDecode($options, $headers, $stream);
  96. try {
  97. $stream = Psr7\stream_for($stream);
  98. } catch (\Throwable $e) {
  99. $stream = Psr7\Utils::streamFor($stream);
  100. }
  101. $sink = $stream;
  102. if (strcasecmp('HEAD', $request->getMethod())) {
  103. $sink = $this->createSink($stream, $options);
  104. }
  105. $response = new Psr7\Response($status, $headers, $sink, $ver, $reason);
  106. if (isset($options['on_headers'])) {
  107. try {
  108. $options['on_headers']($response);
  109. } catch (\Exception $e) {
  110. $msg = 'An error was encountered during the on_headers event';
  111. $ex = new RequestException($msg, $request, $response, $e);
  112. return \GuzzleHttp\Promise\rejection_for($ex);
  113. }
  114. }
  115. if ($sink !== $stream) {
  116. $this->drain(
  117. $stream,
  118. $sink,
  119. $response->getHeaderLine('Content-Length')
  120. );
  121. }
  122. $this->invokeStats($options, $request, $startTime, $response, null);
  123. return new FulfilledPromise($response);
  124. }
  125. private function createSink(StreamInterface $stream, array $options)
  126. {
  127. if (!empty($options['stream'])) {
  128. return $stream;
  129. }
  130. $sink = isset($options['sink'])
  131. ? $options['sink']
  132. : fopen('php://temp', 'r+');
  133. if (is_string($sink)) {
  134. return new Psr7\LazyOpenStream($sink, 'w+');
  135. }
  136. try {
  137. return Psr7\stream_for($sink);
  138. } catch (\Throwable $e) {
  139. return Psr7\Utils::streamFor($sink);
  140. }
  141. }
  142. private function checkDecode(array $options, array $headers, $stream)
  143. {
  144. if (!empty($options['decode_content'])) {
  145. $normalizedKeys = \GuzzleHttp\normalize_header_keys($headers);
  146. if (isset($normalizedKeys['content-encoding'])) {
  147. $encoding = $headers[$normalizedKeys['content-encoding']];
  148. if ($encoding[0] === 'gzip' || $encoding[0] === 'deflate') {
  149. try {
  150. $stream = new Psr7\InflateStream(
  151. Psr7\stream_for($stream)
  152. );
  153. } catch (\Throwable $th) {
  154. $stream = new Psr7\InflateStream(
  155. Psr7\Utils::streamFor($stream)
  156. );
  157. }
  158. $headers['x-encoded-content-encoding']
  159. = $headers[$normalizedKeys['content-encoding']];
  160. unset($headers[$normalizedKeys['content-encoding']]);
  161. if (isset($normalizedKeys['content-length'])) {
  162. $headers['x-encoded-content-length']
  163. = $headers[$normalizedKeys['content-length']];
  164. $length = (int) $stream->getSize();
  165. if ($length === 0) {
  166. unset($headers[$normalizedKeys['content-length']]);
  167. } else {
  168. $headers[$normalizedKeys['content-length']] = [$length];
  169. }
  170. }
  171. }
  172. }
  173. }
  174. return [$stream, $headers];
  175. }
  176. private function drain(
  177. StreamInterface $source,
  178. StreamInterface $sink,
  179. $contentLength
  180. ) {
  181. Psr7\copy_to_stream(
  182. $source,
  183. $sink,
  184. (strlen($contentLength) > 0 && (int) $contentLength > 0) ? (int) $contentLength : -1
  185. );
  186. $sink->seek(0);
  187. $source->close();
  188. return $sink;
  189. }
  190. private function createResource(callable $callback)
  191. {
  192. $errors = null;
  193. set_error_handler(function ($_, $msg, $file, $line) use (&$errors) {
  194. $errors[] = [
  195. 'message' => $msg,
  196. 'file' => $file,
  197. 'line' => $line
  198. ];
  199. return true;
  200. });
  201. $resource = $callback();
  202. restore_error_handler();
  203. if (!$resource) {
  204. $message = 'Error creating resource: ';
  205. foreach ($errors as $err) {
  206. foreach ($err as $key => $value) {
  207. $message .= "[$key] $value" . PHP_EOL;
  208. }
  209. }
  210. throw new \UnexpectedValueException(trim($message));
  211. }
  212. return $resource;
  213. }
  214. private function createStream(RequestInterface $request, array $options)
  215. {
  216. static $methods;
  217. if (!$methods) {
  218. $methods = array_flip(get_class_methods(__CLASS__));
  219. }
  220. if ($request->getProtocolVersion() == '1.1'
  221. && !$request->hasHeader('Connection')
  222. ) {
  223. $request = $request->withHeader('Connection', 'close');
  224. }
  225. if (!isset($options['verify'])) {
  226. $options['verify'] = true;
  227. }
  228. $params = [];
  229. $context = $this->getDefaultContext($request, $options);
  230. if (isset($options['on_headers']) && !is_callable($options['on_headers'])) {
  231. throw new \InvalidArgumentException('on_headers must be callable');
  232. }
  233. if (!empty($options)) {
  234. foreach ($options as $key => $value) {
  235. $method = "add_{$key}";
  236. if (isset($methods[$method])) {
  237. $this->{$method}($request, $context, $value, $params);
  238. }
  239. }
  240. }
  241. if (isset($options['stream_context'])) {
  242. if (!is_array($options['stream_context'])) {
  243. throw new \InvalidArgumentException('stream_context must be an array');
  244. }
  245. $context = array_replace_recursive(
  246. $context,
  247. $options['stream_context']
  248. );
  249. }
  250. if (isset($options['auth'])
  251. && is_array($options['auth'])
  252. && isset($options['auth'][2])
  253. && 'ntlm' == $options['auth'][2]
  254. ) {
  255. throw new \InvalidArgumentException('Microsoft NTLM authentication only supported with curl handler');
  256. }
  257. $uri = $this->resolveHost($request, $options);
  258. $context = $this->createResource(
  259. function () use ($context, $params) {
  260. return stream_context_create($context, $params);
  261. }
  262. );
  263. return $this->createResource(
  264. function () use ($uri, &$http_response_header, $context, $options) {
  265. $resource = fopen((string) $uri, 'r', false, $context);
  266. $this->lastHeaders = $http_response_header;
  267. if (isset($options['read_timeout'])) {
  268. $readTimeout = $options['read_timeout'];
  269. $sec = (int) $readTimeout;
  270. $usec = ($readTimeout - $sec) * 100000;
  271. stream_set_timeout($resource, $sec, $usec);
  272. }
  273. return $resource;
  274. }
  275. );
  276. }
  277. private function resolveHost(RequestInterface $request, array $options)
  278. {
  279. $uri = $request->getUri();
  280. if (isset($options['force_ip_resolve']) && !filter_var($uri->getHost(), FILTER_VALIDATE_IP)) {
  281. if ('v4' === $options['force_ip_resolve']) {
  282. $records = dns_get_record($uri->getHost(), DNS_A);
  283. if (!isset($records[0]['ip'])) {
  284. throw new ConnectException(
  285. sprintf("Could not resolve IPv4 address for host '%s'", $uri->getHost()),
  286. $request
  287. );
  288. }
  289. $uri = $uri->withHost($records[0]['ip']);
  290. }
  291. if ('v6' === $options['force_ip_resolve']) {
  292. $records = dns_get_record($uri->getHost(), DNS_AAAA);
  293. if (!isset($records[0]['ipv6'])) {
  294. throw new ConnectException(
  295. sprintf("Could not resolve IPv6 address for host '%s'", $uri->getHost()),
  296. $request
  297. );
  298. }
  299. $uri = $uri->withHost('[' . $records[0]['ipv6'] . ']');
  300. }
  301. }
  302. return $uri;
  303. }
  304. private function getDefaultContext(RequestInterface $request)
  305. {
  306. $headers = '';
  307. foreach ($request->getHeaders() as $name => $value) {
  308. foreach ($value as $val) {
  309. $headers .= "$name: $val\r\n";
  310. }
  311. }
  312. $context = [
  313. 'http' => [
  314. 'method' => $request->getMethod(),
  315. 'header' => $headers,
  316. 'protocol_version' => $request->getProtocolVersion(),
  317. 'ignore_errors' => true,
  318. 'follow_location' => 0,
  319. ],
  320. ];
  321. $body = (string) $request->getBody();
  322. if (!empty($body)) {
  323. $context['http']['content'] = $body;
  324. if (!$request->hasHeader('Content-Type')) {
  325. $context['http']['header'] .= "Content-Type:\r\n";
  326. }
  327. }
  328. $context['http']['header'] = rtrim($context['http']['header']);
  329. return $context;
  330. }
  331. }