4 * Class to handle concurrent HTTP requests
6 * HTTP request maps use the following format:
7 * - method : GET/HEAD/PUT/POST/DELETE
8 * - url : HTTP/HTTPS URL
9 * - query : <query parameter field/value associative array> (uses RFC 3986)
10 * - headers : <header name/value associative array>
11 * - body : source to get the HTTP request body from;
12 * this can simply be a string (always), a resource for
13 * PUT requests, and a field/value array for POST request;
14 * array bodies are encoded as multipart/form-data and strings
15 * use application/x-www-form-urlencoded (headers sent automatically)
16 * - stream : resource to stream the HTTP response body to
18 * @author Aaron Schulz
21 class MultiHttpClient
{
23 protected $multiHandle = null; // curl_multi handle
24 /** @var string|null SSL certificates path */
25 protected $caBundlePath;
27 protected $connTimeout = 10;
29 protected $reqTimeout = 300;
31 protected $usePipelining = false;
33 protected $maxConnsPerHost = 50;
36 * @param array $options
37 * - connTimeout : default connection timeout
38 * - reqTimeout : default request timeout
39 * - usePipelining : whether to use HTTP pipelining if possible (for all hosts)
40 * - maxConnsPerHost : maximum number of concurrent connections (per host)
42 public function __construct( array $options ) {
43 if ( isset( $options['caBundlePath'] ) ) {
44 $this->caBundlePath
= $options['caBundlePath'];
45 if ( !file_exists( $this->caBundlePath
) ) {
46 throw new Exception( "Cannot find CA bundle: " . $this->caBundlePath
);
49 static $opts = array( 'connTimeout', 'reqTimeout', 'usePipelining', 'maxConnsPerHost' );
50 foreach ( $opts as $key ) {
51 if ( isset( $options[$key] ) ) {
52 $this->$key = $options[$key];
58 * Execute an HTTP(S) request
60 * This method returns a response map of:
61 * - code : HTTP response code or 0 if there was a serious cURL error
62 * - reason : HTTP response reason (empty if there was a serious cURL error)
63 * - headers : <header name/value associative array>
64 * - body : HTTP response body or resource (if "stream" was set)
65 * - err : Any cURL error string
66 * The map also stores integer-indexed copies of these values. This lets callers do:
68 * list( $rcode, $rdesc, $rhdrs, $rbody, $rerr ) = $req;
70 * @param array $req HTTP request array
72 * - connTimeout : connection timeout per request
73 * - reqTimeout : post-connection timeout per request
74 * @return array Response array for request
76 public function run( array $req, array $opts = array() ) {
77 $req = $this->runMulti( array( $req ), $opts );
78 return $req[0]['response'];
82 * Execute a set of HTTP(S) request concurrently
84 * The maps are returned by this method with the 'response' field set to a map of:
85 * - code : HTTP response code or 0 if there was a serious cURL error
86 * - reason : HTTP response reason (empty if there was a serious cURL error)
87 * - headers : <header name/value associative array>
88 * - body : HTTP response body or resource (if "stream" was set)
89 * - err : Any cURL error string
90 * The map also stores integer-indexed copies of these values. This lets callers do:
92 * list( $rcode, $rdesc, $rhdrs, $rbody, $rerr ) = $req;
94 * All headers in the 'headers' field are normalized to use lower case names.
95 * This is true for the request headers and the response headers.
97 * @param array $req Map of HTTP request arrays
99 * - connTimeout : connection timeout per request
100 * - reqTimeout : post-connection timeout per request
101 * - usePipelining : whether to use HTTP pipelining if possible
102 * - maxConnsPerHost : maximum number of concurrent connections (per host)
103 * @return array $reqs With response array populated for each
105 public function runMulti( array $reqs, array $opts = array() ) {
106 $chm = $this->getCurlMulti();
108 // Normalize $reqs and add all of the required cURL handles...
110 foreach ( $reqs as $index => &$req ) {
111 $req['response'] = array(
114 'headers' => array(),
118 if ( !isset( $req['method'] ) ) {
119 throw new Exception( "Request has no 'method' field set." );
120 } elseif ( !isset( $req['url'] ) ) {
121 throw new Exception( "Request has no 'url' field set." );
123 $req['query'] = isset( $req['query'] ) ?
$req['query'] : array();
124 $headers = array(); // normalized headers
125 if ( isset( $req['headers'] ) ) {
126 foreach ( $req['headers'] as $name => $value ) {
127 $headers[strtolower( $name )] = $value;
130 $req['headers'] = $headers;
131 if ( !isset( $req['body'] ) ) {
133 $req['headers']['content-length'] = 0;
135 $handles[$index] = $this->getCurlHandle( $req, $opts );
136 if ( count( $reqs ) > 1 ) {
137 // https://github.com/guzzle/guzzle/issues/349
138 curl_setopt( $handles[$index], CURLOPT_FORBID_REUSE
, true );
142 $indexes = array_keys( $reqs );
143 if ( function_exists( 'curl_multi_setopt' ) ) { // PHP 5.5
144 if ( isset( $opts['usePipelining'] ) ) {
145 curl_multi_setopt( $chm, CURLMOPT_PIPELINING
, (int)$opts['usePipelining'] );
147 if ( isset( $opts['maxConnsPerHost'] ) ) {
148 // Keep these sockets around as they may be needed later in the request
149 curl_multi_setopt( $chm, CURLMOPT_MAXCONNECTS
, (int)$opts['maxConnsPerHost'] );
153 // @TODO: use a per-host rolling handle window (e.g. CURLMOPT_MAX_HOST_CONNECTIONS)
154 $batches = array_chunk( $indexes, $this->maxConnsPerHost
);
156 foreach ( $batches as $batch ) {
157 // Attach all cURL handles for this batch
158 foreach ( $batch as $index ) {
159 curl_multi_add_handle( $chm, $handles[$index] );
161 // Execute the cURL handles concurrently...
162 $active = null; // handles still being processed
164 // Do any available work...
166 $mrc = curl_multi_exec( $chm, $active );
167 } while ( $mrc == CURLM_CALL_MULTI_PERFORM
);
168 // Wait (if possible) for available work...
169 if ( $active > 0 && $mrc == CURLM_OK
) {
170 if ( curl_multi_select( $chm, 10 ) == -1 ) {
171 // PHP bug 63411; http://curl.haxx.se/libcurl/c/curl_multi_fdset.html
172 usleep( 5000 ); // 5ms
175 } while ( $active > 0 && $mrc == CURLM_OK
);
178 // Remove all of the added cURL handles and check for errors...
179 foreach ( $reqs as $index => &$req ) {
180 $ch = $handles[$index];
181 curl_multi_remove_handle( $chm, $ch );
182 if ( curl_errno( $ch ) !== 0 ) {
183 $req['error'] = "(curl error: " . curl_errno( $ch ) . ") " . curl_error( $ch );
185 // For convenience with the list() operator
186 $req['response'][0] = $req['response']['code'];
187 $req['response'][1] = $req['response']['reason'];
188 $req['response'][2] = $req['response']['headers'];
189 $req['response'][3] = $req['response']['body'];
190 $req['response'][4] = $req['response']['error'];
192 // Close any string wrapper file handles
193 if ( isset( $req['_closeHandle'] ) ) {
194 fclose( $req['_closeHandle'] );
195 unset( $req['_closeHandle'] );
199 // Restore the default settings
200 if ( function_exists( 'curl_multi_setopt' ) ) { // PHP 5.5
201 curl_multi_setopt( $chm, CURLMOPT_PIPELINING
, (int)$this->usePipelining
);
202 curl_multi_setopt( $chm, CURLMOPT_MAXCONNECTS
, (int)$this->maxConnsPerHost
);
209 * @param array $req HTTP request map
211 * - connTimeout : default connection timeout
212 * - reqTimeout : default request timeout
215 protected function getCurlHandle( array &$req, array $opts = array() ) {
218 curl_setopt( $ch, CURLOPT_CONNECTTIMEOUT
,
219 isset( $opts['connTimeout'] ) ?
$opts['connTimeout'] : $this->connTimeout
);
220 curl_setopt( $ch, CURLOPT_TIMEOUT
,
221 isset( $opts['reqTimeout'] ) ?
$opts['reqTimeout'] : $this->reqTimeout
);
222 curl_setopt( $ch, CURLOPT_FOLLOWLOCATION
, 1 );
223 curl_setopt( $ch, CURLOPT_MAXREDIRS
, 4 );
224 curl_setopt( $ch, CURLOPT_HEADER
, 0 );
225 if ( !is_null( $this->caBundlePath
) ) {
226 curl_setopt( $ch, CURLOPT_SSL_VERIFYPEER
, true );
227 curl_setopt( $ch, CURLOPT_CAINFO
, $this->caBundlePath
);
229 curl_setopt( $ch, CURLOPT_RETURNTRANSFER
, 1 );
232 // PHP_QUERY_RFC3986 is PHP 5.4+ only
233 $query = str_replace(
236 http_build_query( $req['query'], '', '&' )
238 if ( $query != '' ) {
239 $url .= strpos( $req['url'], '?' ) === false ?
"?$query" : "&$query";
241 curl_setopt( $ch, CURLOPT_URL
, $url );
243 curl_setopt( $ch, CURLOPT_CUSTOMREQUEST
, $req['method'] );
244 if ( $req['method'] === 'HEAD' ) {
245 curl_setopt( $ch, CURLOPT_NOBODY
, 1 );
248 if ( $req['method'] === 'PUT' ) {
249 curl_setopt( $ch, CURLOPT_PUT
, 1 );
250 if ( is_resource( $req['body'] ) ) {
251 curl_setopt( $ch, CURLOPT_INFILE
, $req['body'] );
252 if ( isset( $req['headers']['content-length'] ) ) {
253 curl_setopt( $ch, CURLOPT_INFILESIZE
, $req['headers']['content-length'] );
254 } elseif ( isset( $req['headers']['transfer-encoding'] ) &&
255 $req['headers']['transfer-encoding'] === 'chunks'
257 curl_setopt( $ch, CURLOPT_UPLOAD
, true );
259 throw new Exception( "Missing 'Content-Length' or 'Transfer-Encoding' header." );
261 } elseif ( $req['body'] !== '' ) {
262 $fp = fopen( "php://temp", "wb+" );
263 fwrite( $fp, $req['body'], strlen( $req['body'] ) );
265 curl_setopt( $ch, CURLOPT_INFILE
, $fp );
266 curl_setopt( $ch, CURLOPT_INFILESIZE
, strlen( $req['body'] ) );
267 $req['_closeHandle'] = $fp; // remember to close this later
269 curl_setopt( $ch, CURLOPT_INFILESIZE
, 0 );
271 curl_setopt( $ch, CURLOPT_READFUNCTION
,
272 function ( $ch, $fd, $length ) {
273 $data = fread( $fd, $length );
274 $len = strlen( $data );
278 } elseif ( $req['method'] === 'POST' ) {
279 curl_setopt( $ch, CURLOPT_POST
, 1 );
280 curl_setopt( $ch, CURLOPT_POSTFIELDS
, $req['body'] );
282 if ( is_resource( $req['body'] ) ||
$req['body'] !== '' ) {
283 throw new Exception( "HTTP body specified for a non PUT/POST request." );
285 $req['headers']['content-length'] = 0;
289 foreach ( $req['headers'] as $name => $value ) {
290 if ( strpos( $name, ': ' ) ) {
291 throw new Exception( "Headers cannot have ':' in the name." );
293 $headers[] = $name . ': ' . trim( $value );
295 curl_setopt( $ch, CURLOPT_HTTPHEADER
, $headers );
297 curl_setopt( $ch, CURLOPT_HEADERFUNCTION
,
298 function ( $ch, $header ) use ( &$req ) {
299 $length = strlen( $header );
301 if ( preg_match( "/^(HTTP\/1\.[01]) (\d{3}) (.*)/", $header, $matches ) ) {
302 $req['response']['code'] = (int)$matches[2];
303 $req['response']['reason'] = trim( $matches[3] );
306 if ( strpos( $header, ":" ) === false ) {
309 list( $name, $value ) = explode( ":", $header, 2 );
310 $req['response']['headers'][strtolower( $name )] = trim( $value );
315 if ( isset( $req['stream'] ) ) {
316 // Don't just use CURLOPT_FILE as that might give:
317 // curl_setopt(): cannot represent a stream of type Output as a STDIO FILE*
318 // The callback here handles both normal files and php://temp handles.
319 curl_setopt( $ch, CURLOPT_WRITEFUNCTION
,
320 function ( $ch, $data ) use ( &$req ) {
321 return fwrite( $req['stream'], $data );
325 curl_setopt( $ch, CURLOPT_WRITEFUNCTION
,
326 function ( $ch, $data ) use ( &$req ) {
327 $req['response']['body'] .= $data;
328 return strlen( $data );
339 protected function getCurlMulti() {
340 if ( !$this->multiHandle
) {
341 $cmh = curl_multi_init();
342 if ( function_exists( 'curl_multi_setopt' ) ) { // PHP 5.5
343 curl_multi_setopt( $cmh, CURLMOPT_PIPELINING
, (int)$this->usePipelining
);
344 curl_multi_setopt( $cmh, CURLMOPT_MAXCONNECTS
, (int)$this->maxConnsPerHost
);
346 $this->multiHandle
= $cmh;
348 return $this->multiHandle
;
351 function __destruct() {
352 if ( $this->multiHandle
) {
353 curl_multi_close( $this->multiHandle
);