scratch – Rev 87
?pathlinks?
<?php
namespace GuzzleHttp\Adapter\Curl;
use GuzzleHttp\Adapter\AdapterInterface;
use GuzzleHttp\Adapter\ParallelAdapterInterface;
use GuzzleHttp\Adapter\TransactionInterface;
use GuzzleHttp\Event\RequestEvents;
use GuzzleHttp\Exception\AdapterException;
use GuzzleHttp\Exception\RequestException;
use GuzzleHttp\Message\MessageFactoryInterface;
/**
* HTTP adapter that uses cURL multi as a transport layer
*
* When using the CurlAdapter, custom curl options can be specified as an
* associative array of curl option constants mapping to values in the
* **curl** key of a request's configuration options.
*
* In addition to being able to supply configuration options via the curl
* request config, you can also specify the select_timeout variable using the
* `GUZZLE_CURL_SELECT_TIMEOUT` environment variable.
*/
class MultiAdapter implements AdapterInterface, ParallelAdapterInterface
{
const ERROR_STR = 'See http://curl.haxx.se/libcurl/c/libcurl-errors.html for an explanation of cURL errors';
const ENV_SELECT_TIMEOUT = 'GUZZLE_CURL_SELECT_TIMEOUT';
/** @var CurlFactory */
private $curlFactory;
/** @var MessageFactoryInterface */
private $messageFactory;
/** @var array Array of curl multi handles */
private $multiHandles = [];
/** @var array Array of curl multi handles */
private $multiOwned = [];
/** @var int Total number of idle handles to keep in cache */
private $maxHandles;
/** @var double */
private $selectTimeout;
/**
* Accepts an associative array of options:
*
* - handle_factory: Optional callable factory used to create cURL handles.
* The callable is invoked with the following arguments:
* TransactionInterface, MessageFactoryInterface, and an optional cURL
* handle to modify. The factory method must then return a cURL resource.
* - select_timeout: Specify a float in seconds to use for a
* curl_multi_select timeout.
* - max_handles: Maximum number of idle handles (defaults to 3).
*
* @param MessageFactoryInterface $messageFactory
* @param array $options Array of options to use with the adapter:
*/
public function __construct(
MessageFactoryInterface $messageFactory,
array $options = []
) {
$this->messageFactory = $messageFactory;
$this->curlFactory = isset($options['handle_factory'])
? $options['handle_factory']
: new CurlFactory();
if (isset($options['select_timeout'])) {
$this->selectTimeout = $options['select_timeout'];
} elseif (isset($_SERVER[self::ENV_SELECT_TIMEOUT])) {
$this->selectTimeout = $_SERVER[self::ENV_SELECT_TIMEOUT];
} else {
$this->selectTimeout = 1;
}
$this->maxHandles = isset($options['max_handles'])
? $options['max_handles']
: 3;
}
public function __destruct()
{
foreach ($this->multiHandles as $handle) {
if (is_resource($handle)) {
curl_multi_close($handle);
}
}
}
/**
* Throw an exception for a cURL multi response
*
* @param int $code Curl response code
* @throws AdapterException
*/
public static function throwMultiError($code)
{
$buffer = function_exists('curl_multi_strerror')
? curl_multi_strerror($code)
: self::ERROR_STR;
throw new AdapterException(sprintf('cURL error %s: %s', $code, $buffer));
}
public function send(TransactionInterface $transaction)
{
$context = new BatchContext($this->checkoutMultiHandle(), true);
$this->addHandle($transaction, $context);
$this->perform($context);
return $transaction->getResponse();
}
public function sendAll(\Iterator $transactions, $parallel)
{
$context = new BatchContext(
$this->checkoutMultiHandle(),
false,
$transactions
);
foreach (new \LimitIterator($transactions, 0, $parallel) as $trans) {
$this->addHandle($trans, $context);
}
$this->perform($context);
}
private function perform(BatchContext $context)
{
// The first curl_multi_select often times out no matter what, but is
// usually required for fast transfers.
$active = false;
$multi = $context->getMultiHandle();
do {
do {
$mrc = curl_multi_exec($multi, $active);
} while ($mrc === CURLM_CALL_MULTI_PERFORM);
if ($mrc != CURLM_OK) {
self::throwMultiError($mrc);
}
$this->processMessages($context);
if ($active &&
curl_multi_select($multi, $this->selectTimeout) === -1
) {
// Perform a usleep if a select returns -1.
// See: https://bugs.php.net/bug.php?id=61141
usleep(250);
}
} while ($context->isActive() || $active);
$this->releaseMultiHandle($multi, $this->maxHandles);
}
private function processMessages(BatchContext $context)
{
$multi = $context->getMultiHandle();
while ($done = curl_multi_info_read($multi)) {
$transaction = $context->findTransaction($done['handle']);
$this->processResponse($transaction, $done, $context);
// Add the next transaction if there are more in the queue
if ($next = $context->nextPending()) {
$this->addHandle($next, $context);
}
}
}
private function processResponse(
TransactionInterface $transaction,
array $curl,
BatchContext $context
) {
$info = $context->removeTransaction($transaction);
try {
if (!$this->isCurlException($transaction, $curl, $context, $info) &&
$this->validateResponseWasSet($transaction, $context)
) {
if ($body = $transaction->getResponse()->getBody()) {
$body->seek(0);
}
RequestEvents::emitComplete($transaction, $info);
}
} catch (\Exception $e) {
$this->throwException($e, $context);
}
}
private function addHandle(
TransactionInterface $transaction,
BatchContext $context
) {
try {
RequestEvents::emitBefore($transaction);
// Only transfer if the request was not intercepted
if (!$transaction->getResponse()) {
$factory = $this->curlFactory;
$context->addTransaction(
$transaction,
$factory($transaction, $this->messageFactory)
);
}
} catch (RequestException $e) {
$this->throwException($e, $context);
}
}
private function isCurlException(
TransactionInterface $transaction,
array $curl,
BatchContext $context,
array $info
) {
if (CURLM_OK == $curl['result'] ||
CURLM_CALL_MULTI_PERFORM == $curl['result']
) {
return false;
}
$request = $transaction->getRequest();
try {
// Send curl stats along if they are available
$stats = ['curl_result' => $curl['result']] + $info;
RequestEvents::emitError(
$transaction,
new RequestException(
sprintf(
'[curl] (#%s) %s [url] %s',
$curl['result'],
function_exists('curl_strerror')
? curl_strerror($curl['result'])
: self::ERROR_STR,
$request->getUrl()
),
$request
),
$stats
);
} catch (\Exception $e) {
$this->throwException($e, $context);
}
return true;
}
private function throwException(\Exception $e, BatchContext $context)
{
if ($context->throwsExceptions()
|| ($e instanceof RequestException && $e->getThrowImmediately())
) {
$context->removeAll();
$this->releaseMultiHandle($context->getMultiHandle(), -1);
throw $e;
}
}
/**
* Returns a curl_multi handle from the cache or creates a new one
*
* @return resource
*/
private function checkoutMultiHandle()
{
// Find an unused handle in the cache
$key = array_search(false, $this->multiOwned, true);
if (false !== $key) {
$this->multiOwned[$key] = true;
return $this->multiHandles[$key];
}
// Add a new handle
$handle = curl_multi_init();
$id = (int) $handle;
$this->multiHandles[$id] = $handle;
$this->multiOwned[$id] = true;
return $handle;
}
/**
* Releases a curl_multi handle back into the cache and removes excess cache
*
* @param resource $handle Curl multi handle to remove
* @param int $maxHandles (Optional) Maximum number of existing multiHandles to allow before pruning.
*/
private function releaseMultiHandle($handle, $maxHandles)
{
$id = (int) $handle;
if (count($this->multiHandles) <= $maxHandles) {
$this->multiOwned[$id] = false;
} elseif (isset($this->multiHandles[$id], $this->multiOwned[$id])) {
// Prune excessive handles
curl_multi_close($this->multiHandles[$id]);
unset($this->multiHandles[$id], $this->multiOwned[$id]);
}
}
/**
* This function ensures that a response was set on a transaction. If one
* was not set, then the request is retried if possible. This error
* typically means you are sending a payload, curl encountered a
* "Connection died, retrying a fresh connect" error, tried to rewind the
* stream, and then encountered a "necessary data rewind wasn't possible"
* error, causing the request to be sent through curl_multi_info_read()
* without an error status.
*
* @param TransactionInterface $transaction
* @param BatchContext $context
*
* @return bool Returns true if it's OK, and false if it failed.
* @throws \GuzzleHttp\Exception\RequestException If it failed and cannot
* recover.
*/
private function validateResponseWasSet(
TransactionInterface $transaction,
BatchContext $context
) {
if ($transaction->getResponse()) {
return true;
}
$body = $transaction->getRequest()->getBody();
if (!$body) {
// This is weird and should probably never happen.
RequestEvents::emitError(
$transaction,
new RequestException(
'No response was received for a request with no body. This'
. ' could mean that you are saturating your network.',
$transaction->getRequest()
)
);
} elseif (!$body->isSeekable() || !$body->seek(0)) {
// Nothing we can do with this. Sorry!
RequestEvents::emitError(
$transaction,
new RequestException(
'The connection was unexpectedly closed. The request would'
. ' have been retried, but attempting to rewind the'
. ' request body failed. Consider wrapping your request'
. ' body in a CachingStream decorator to work around this'
. ' issue if necessary.',
$transaction->getRequest()
)
);
} else {
$this->retryFailedConnection($transaction, $context);
}
return false;
}
private function retryFailedConnection(
TransactionInterface $transaction,
BatchContext $context
) {
// Add the request back to the batch to retry automatically.
$context->addTransaction(
$transaction,
call_user_func(
$this->curlFactory,
$transaction,
$this->messageFactory
)
);
}
}