annotate vendor/guzzlehttp/guzzle/src/Handler/CurlMultiHandler.php @ 19:fa3358dc1485 tip

Add ndrum files
author Chris Cannam
date Wed, 28 Aug 2019 13:14:47 +0100
parents 5fb285c0d0e3
children
rev   line source
Chris@0 1 <?php
Chris@0 2 namespace GuzzleHttp\Handler;
Chris@0 3
Chris@0 4 use GuzzleHttp\Promise as P;
Chris@0 5 use GuzzleHttp\Promise\Promise;
Chris@0 6 use GuzzleHttp\Psr7;
Chris@0 7 use Psr\Http\Message\RequestInterface;
Chris@0 8
Chris@0 9 /**
Chris@0 10 * Returns an asynchronous response using curl_multi_* functions.
Chris@0 11 *
Chris@0 12 * When using the CurlMultiHandler, custom curl options can be specified as an
Chris@0 13 * associative array of curl option constants mapping to values in the
Chris@0 14 * **curl** key of the provided request options.
Chris@0 15 *
Chris@0 16 * @property resource $_mh Internal use only. Lazy loaded multi-handle.
Chris@0 17 */
Chris@0 18 class CurlMultiHandler
Chris@0 19 {
Chris@0 20 /** @var CurlFactoryInterface */
Chris@0 21 private $factory;
Chris@0 22 private $selectTimeout;
Chris@0 23 private $active;
Chris@0 24 private $handles = [];
Chris@0 25 private $delays = [];
Chris@0 26
Chris@0 27 /**
Chris@0 28 * This handler accepts the following options:
Chris@0 29 *
Chris@0 30 * - handle_factory: An optional factory used to create curl handles
Chris@0 31 * - select_timeout: Optional timeout (in seconds) to block before timing
Chris@0 32 * out while selecting curl handles. Defaults to 1 second.
Chris@0 33 *
Chris@0 34 * @param array $options
Chris@0 35 */
Chris@0 36 public function __construct(array $options = [])
Chris@0 37 {
Chris@0 38 $this->factory = isset($options['handle_factory'])
Chris@0 39 ? $options['handle_factory'] : new CurlFactory(50);
Chris@0 40 $this->selectTimeout = isset($options['select_timeout'])
Chris@0 41 ? $options['select_timeout'] : 1;
Chris@0 42 }
Chris@0 43
Chris@0 44 public function __get($name)
Chris@0 45 {
Chris@0 46 if ($name === '_mh') {
Chris@0 47 return $this->_mh = curl_multi_init();
Chris@0 48 }
Chris@0 49
Chris@0 50 throw new \BadMethodCallException();
Chris@0 51 }
Chris@0 52
Chris@0 53 public function __destruct()
Chris@0 54 {
Chris@0 55 if (isset($this->_mh)) {
Chris@0 56 curl_multi_close($this->_mh);
Chris@0 57 unset($this->_mh);
Chris@0 58 }
Chris@0 59 }
Chris@0 60
Chris@0 61 public function __invoke(RequestInterface $request, array $options)
Chris@0 62 {
Chris@0 63 $easy = $this->factory->create($request, $options);
Chris@0 64 $id = (int) $easy->handle;
Chris@0 65
Chris@0 66 $promise = new Promise(
Chris@0 67 [$this, 'execute'],
Chris@13 68 function () use ($id) {
Chris@13 69 return $this->cancel($id);
Chris@13 70 }
Chris@0 71 );
Chris@0 72
Chris@0 73 $this->addRequest(['easy' => $easy, 'deferred' => $promise]);
Chris@0 74
Chris@0 75 return $promise;
Chris@0 76 }
Chris@0 77
Chris@0 78 /**
Chris@0 79 * Ticks the curl event loop.
Chris@0 80 */
Chris@0 81 public function tick()
Chris@0 82 {
Chris@0 83 // Add any delayed handles if needed.
Chris@0 84 if ($this->delays) {
Chris@0 85 $currentTime = microtime(true);
Chris@0 86 foreach ($this->delays as $id => $delay) {
Chris@0 87 if ($currentTime >= $delay) {
Chris@0 88 unset($this->delays[$id]);
Chris@0 89 curl_multi_add_handle(
Chris@0 90 $this->_mh,
Chris@0 91 $this->handles[$id]['easy']->handle
Chris@0 92 );
Chris@0 93 }
Chris@0 94 }
Chris@0 95 }
Chris@0 96
Chris@0 97 // Step through the task queue which may add additional requests.
Chris@0 98 P\queue()->run();
Chris@0 99
Chris@0 100 if ($this->active &&
Chris@0 101 curl_multi_select($this->_mh, $this->selectTimeout) === -1
Chris@0 102 ) {
Chris@0 103 // Perform a usleep if a select returns -1.
Chris@0 104 // See: https://bugs.php.net/bug.php?id=61141
Chris@0 105 usleep(250);
Chris@0 106 }
Chris@0 107
Chris@0 108 while (curl_multi_exec($this->_mh, $this->active) === CURLM_CALL_MULTI_PERFORM);
Chris@0 109
Chris@0 110 $this->processMessages();
Chris@0 111 }
Chris@0 112
Chris@0 113 /**
Chris@0 114 * Runs until all outstanding connections have completed.
Chris@0 115 */
Chris@0 116 public function execute()
Chris@0 117 {
Chris@0 118 $queue = P\queue();
Chris@0 119
Chris@0 120 while ($this->handles || !$queue->isEmpty()) {
Chris@0 121 // If there are no transfers, then sleep for the next delay
Chris@0 122 if (!$this->active && $this->delays) {
Chris@0 123 usleep($this->timeToNext());
Chris@0 124 }
Chris@0 125 $this->tick();
Chris@0 126 }
Chris@0 127 }
Chris@0 128
Chris@0 129 private function addRequest(array $entry)
Chris@0 130 {
Chris@0 131 $easy = $entry['easy'];
Chris@0 132 $id = (int) $easy->handle;
Chris@0 133 $this->handles[$id] = $entry;
Chris@0 134 if (empty($easy->options['delay'])) {
Chris@0 135 curl_multi_add_handle($this->_mh, $easy->handle);
Chris@0 136 } else {
Chris@0 137 $this->delays[$id] = microtime(true) + ($easy->options['delay'] / 1000);
Chris@0 138 }
Chris@0 139 }
Chris@0 140
Chris@0 141 /**
Chris@0 142 * Cancels a handle from sending and removes references to it.
Chris@0 143 *
Chris@0 144 * @param int $id Handle ID to cancel and remove.
Chris@0 145 *
Chris@0 146 * @return bool True on success, false on failure.
Chris@0 147 */
Chris@0 148 private function cancel($id)
Chris@0 149 {
Chris@0 150 // Cannot cancel if it has been processed.
Chris@0 151 if (!isset($this->handles[$id])) {
Chris@0 152 return false;
Chris@0 153 }
Chris@0 154
Chris@0 155 $handle = $this->handles[$id]['easy']->handle;
Chris@0 156 unset($this->delays[$id], $this->handles[$id]);
Chris@0 157 curl_multi_remove_handle($this->_mh, $handle);
Chris@0 158 curl_close($handle);
Chris@0 159
Chris@0 160 return true;
Chris@0 161 }
Chris@0 162
Chris@0 163 private function processMessages()
Chris@0 164 {
Chris@0 165 while ($done = curl_multi_info_read($this->_mh)) {
Chris@0 166 $id = (int) $done['handle'];
Chris@0 167 curl_multi_remove_handle($this->_mh, $done['handle']);
Chris@0 168
Chris@0 169 if (!isset($this->handles[$id])) {
Chris@0 170 // Probably was cancelled.
Chris@0 171 continue;
Chris@0 172 }
Chris@0 173
Chris@0 174 $entry = $this->handles[$id];
Chris@0 175 unset($this->handles[$id], $this->delays[$id]);
Chris@0 176 $entry['easy']->errno = $done['result'];
Chris@0 177 $entry['deferred']->resolve(
Chris@0 178 CurlFactory::finish(
Chris@0 179 $this,
Chris@0 180 $entry['easy'],
Chris@0 181 $this->factory
Chris@0 182 )
Chris@0 183 );
Chris@0 184 }
Chris@0 185 }
Chris@0 186
Chris@0 187 private function timeToNext()
Chris@0 188 {
Chris@0 189 $currentTime = microtime(true);
Chris@0 190 $nextTime = PHP_INT_MAX;
Chris@0 191 foreach ($this->delays as $time) {
Chris@0 192 if ($time < $nextTime) {
Chris@0 193 $nextTime = $time;
Chris@0 194 }
Chris@0 195 }
Chris@0 196
Chris@0 197 return max(0, $nextTime - $currentTime) * 1000000;
Chris@0 198 }
Chris@0 199 }