class-curltransport.php
1 week ago
class-sockettransport.php
1 week ago
interface-transportinterface.php
1 week ago
class-sockettransport.php
583 lines
| 1 | <?php |
| 2 | |
| 3 | namespace WordPress\HttpClient\Transport; |
| 4 | |
| 5 | use WordPress\ByteStream\ByteTransformer\InflateTransformer; |
| 6 | use WordPress\ByteStream\ReadStream\FileReadStream; |
| 7 | use WordPress\ByteStream\ReadStream\TransformedReadStream; |
| 8 | use WordPress\HttpClient\ByteStream\ChunkedDecoderReadStream; |
| 9 | use WordPress\HttpClient\ByteStream\ChunkedEncoderByteTransformer; |
| 10 | use WordPress\HttpClient\Client; |
| 11 | use WordPress\HttpClient\ClientState; |
| 12 | use WordPress\HttpClient\HttpError; |
| 13 | use WordPress\HttpClient\Request; |
| 14 | use WordPress\HttpClient\Response; |
| 15 | |
| 16 | /** |
| 17 | * An HTTP client using stream_socket_client(). Supports |
| 18 | * concurrent connections just like curl_multi. |
| 19 | * |
| 20 | * Supports: |
| 21 | * * Concurrency |
| 22 | * * HTTP 1.0 and 1.1 |
| 23 | * * HTTPS via TLS 1.2 and 1.3 |
| 24 | * * Chunked transfer encoding |
| 25 | * * Streaming requests and responses |
| 26 | * * GZip and Deflate transfer encoding |
| 27 | */ |
| 28 | class SocketTransport implements TransportInterface { |
| 29 | |
| 30 | protected const STREAM_SELECT_READ = 1; |
| 31 | protected const STREAM_SELECT_WRITE = 2; |
| 32 | |
| 33 | /** |
| 34 | * @var ClientState |
| 35 | */ |
| 36 | protected $state; |
| 37 | |
| 38 | public function __construct( ClientState $state ) { |
| 39 | $this->state = $state; |
| 40 | } |
| 41 | |
| 42 | public function event_loop_tick(): bool { |
| 43 | if ( 0 === count( $this->state->get_active_requests() ) ) { |
| 44 | return false; |
| 45 | } |
| 46 | |
| 47 | foreach ( $this->state->get_active_requests( |
| 48 | array( |
| 49 | Request::STATE_WILL_ENABLE_CRYPTO, |
| 50 | Request::STATE_WILL_SEND_HEADERS, |
| 51 | Request::STATE_WILL_SEND_BODY, |
| 52 | Request::STATE_SENT, |
| 53 | Request::STATE_RECEIVING_HEADERS, |
| 54 | Request::STATE_RECEIVING_BODY, |
| 55 | Request::STATE_RECEIVED, |
| 56 | ) |
| 57 | ) as $request ) { |
| 58 | $time_elapsed_ms = $this->state->connections[ $request->id ]->time_elapsed_ms(); |
| 59 | if ( $time_elapsed_ms > $this->state->request_timeout_ms ) { |
| 60 | $this->set_error( $request, new HttpError( sprintf( 'Request timed out after %d ms.', (int) $time_elapsed_ms ) ) ); |
| 61 | } |
| 62 | } |
| 63 | |
| 64 | $this->open_nonblocking_http_sockets( |
| 65 | $this->state->get_active_requests( Request::STATE_ENQUEUED ) |
| 66 | ); |
| 67 | |
| 68 | $this->enable_crypto( |
| 69 | $this->state->get_active_requests( Request::STATE_WILL_ENABLE_CRYPTO ) |
| 70 | ); |
| 71 | |
| 72 | $this->send_request_headers( |
| 73 | $this->state->get_active_requests( Request::STATE_WILL_SEND_HEADERS ) |
| 74 | ); |
| 75 | |
| 76 | $this->send_request_body( |
| 77 | $this->state->get_active_requests( Request::STATE_WILL_SEND_BODY ) |
| 78 | ); |
| 79 | |
| 80 | $nb_headers_received = $this->receive_response_headers( |
| 81 | $this->state->get_active_requests( Request::STATE_RECEIVING_HEADERS ) |
| 82 | ); |
| 83 | |
| 84 | foreach ( $this->state->get_active_requests( Request::STATE_RECEIVED ) as $request ) { |
| 85 | $this->mark_finished( $request ); |
| 86 | } |
| 87 | |
| 88 | /** |
| 89 | * Allows the caller to consume the headers before we start polling |
| 90 | * for the body of those requests. |
| 91 | * |
| 92 | * This prevents the following scenario: |
| 93 | * |
| 94 | * 1. The consumer calls await_next_event() and they're only interested in |
| 95 | * the EVENT_GOT_HEADERS event. |
| 96 | * 2. In the same event_loop_tick: |
| 97 | * * The headers arrive |
| 98 | * * The request is promoted to STATE_RECEIVING_BODY |
| 99 | * * We poll for the response body |
| 100 | * * We wait 10 more seconds before the body starts arriving |
| 101 | * 3. The consumer gets the EVENT_GOT_HEADERS event 10 seconds later |
| 102 | * than they could have. |
| 103 | */ |
| 104 | if ( $nb_headers_received > 0 ) { |
| 105 | return true; |
| 106 | } |
| 107 | |
| 108 | $this->receive_response_body( |
| 109 | $this->state->get_active_requests( Request::STATE_RECEIVING_BODY ) |
| 110 | ); |
| 111 | |
| 112 | return true; |
| 113 | } |
| 114 | |
| 115 | /** |
| 116 | * Opens HTTP or HTTPS streams using stream_socket_client() without blocking, |
| 117 | * and returns nearly immediately. |
| 118 | * |
| 119 | * The act of opening a stream is non-blocking itself. This function uses |
| 120 | * a tcp:// stream wrapper, because both https:// and ssl:// wrappers would block |
| 121 | * until the SSL handshake is complete. |
| 122 | * The actual socket it then switched to non-blocking mode using stream_set_blocking(). |
| 123 | * |
| 124 | * @param Request $requests The Request to open the socket for. |
| 125 | * |
| 126 | * @return bool Whether the stream was opened successfully. |
| 127 | */ |
| 128 | protected function open_nonblocking_http_sockets( $requests ) { |
| 129 | foreach ( $requests as $request ) { |
| 130 | $url = $request->url; |
| 131 | $parts = parse_url( $url ); |
| 132 | $scheme = $parts['scheme']; |
| 133 | if ( ! in_array( $scheme, array( 'http', 'https' ), true ) ) { |
| 134 | $this->set_error( |
| 135 | $request, |
| 136 | new HttpError( 'stream_http_open_nonblocking: Invalid scheme in URL ' . $url . ' – only http:// and https:// URLs are supported' ) |
| 137 | ); |
| 138 | continue; |
| 139 | } |
| 140 | |
| 141 | $is_ssl = 'https' === $scheme; |
| 142 | $port = $parts['port'] ?? ( 'https' === $scheme ? 443 : 80 ); |
| 143 | $host = $parts['host']; |
| 144 | |
| 145 | // Create stream context. |
| 146 | $context = stream_context_create( |
| 147 | array( |
| 148 | 'socket' => array( |
| 149 | 'isSsl' => $is_ssl, |
| 150 | 'originalUrl' => $url, |
| 151 | 'socketUrl' => 'tcp://' . $host . ':' . $port, |
| 152 | ), |
| 153 | ) |
| 154 | ); |
| 155 | |
| 156 | // phpcs:ignore WordPress.PHP.NoSilencedErrors.Discouraged |
| 157 | $stream = @stream_socket_client( |
| 158 | 'tcp://' . $host . ':' . $port, |
| 159 | $errno, |
| 160 | $errstr, |
| 161 | $this->state->request_timeout_ms / 1000, |
| 162 | STREAM_CLIENT_CONNECT | STREAM_CLIENT_ASYNC_CONNECT, |
| 163 | $context |
| 164 | ); |
| 165 | |
| 166 | if ( false === $stream ) { |
| 167 | $this->set_error( |
| 168 | $request, |
| 169 | new HttpError( "stream_http_open_nonblocking: stream_socket_client() was unable to open a stream to $url. $errno: $errstr" ) |
| 170 | ); |
| 171 | continue; |
| 172 | } |
| 173 | |
| 174 | stream_set_blocking( $stream, false ); |
| 175 | |
| 176 | $this->state->connections[ $request->id ]->http_socket = $stream; |
| 177 | $this->state->connections[ $request->id ]->started_at = microtime( true ); |
| 178 | if ( $is_ssl ) { |
| 179 | $request->state = Request::STATE_WILL_ENABLE_CRYPTO; |
| 180 | } else { |
| 181 | $request->state = Request::STATE_WILL_SEND_HEADERS; |
| 182 | } |
| 183 | } |
| 184 | |
| 185 | return true; |
| 186 | } |
| 187 | |
| 188 | /** |
| 189 | * Handle transfer encodings. |
| 190 | * |
| 191 | * @param Request $request |
| 192 | * |
| 193 | * @return false|resource |
| 194 | */ |
| 195 | protected function decode_and_monitor_response_body_stream( Request $request ) { |
| 196 | $transfer_encodings = array(); |
| 197 | |
| 198 | $transfer_encoding = $request->response->get_header( 'transfer-encoding' ); |
| 199 | if ( $transfer_encoding ) { |
| 200 | $transfer_encodings = array_map( 'trim', explode( ',', $transfer_encoding ) ); |
| 201 | } |
| 202 | |
| 203 | $content_encoding = $request->response->get_header( 'content-encoding' ); |
| 204 | if ( $content_encoding && ! in_array( $content_encoding, $transfer_encodings, true ) ) { |
| 205 | $transfer_encodings[] = $content_encoding; |
| 206 | } |
| 207 | |
| 208 | $body_stream = FileReadStream::from_resource( |
| 209 | $this->state->connections[ $request->id ]->http_socket |
| 210 | ); |
| 211 | |
| 212 | $transformers = array(); |
| 213 | foreach ( $transfer_encodings as $transfer_encoding ) { |
| 214 | switch ( $transfer_encoding ) { |
| 215 | case 'chunked': |
| 216 | $body_stream = new ChunkedDecoderReadStream( $body_stream ); |
| 217 | break; |
| 218 | case 'gzip': |
| 219 | $transformers[] = new InflateTransformer( |
| 220 | 'gzip' === $transfer_encoding ? ZLIB_ENCODING_GZIP : ZLIB_ENCODING_RAW |
| 221 | ); |
| 222 | break; |
| 223 | case 'deflate': |
| 224 | $transformers[] = new InflateTransformer( ZLIB_ENCODING_DEFLATE ); |
| 225 | break; |
| 226 | case 'identity': |
| 227 | // No-op. |
| 228 | break; |
| 229 | default: |
| 230 | $this->set_error( |
| 231 | $request, |
| 232 | new HttpError( 'Unsupported transfer encoding received from the server: ' . $transfer_encoding ) |
| 233 | ); |
| 234 | break; |
| 235 | } |
| 236 | } |
| 237 | |
| 238 | return new TransformedReadStream( |
| 239 | $body_stream, |
| 240 | $transformers |
| 241 | ); |
| 242 | } |
| 243 | |
| 244 | /** |
| 245 | * Sends HTTP requests using streams. |
| 246 | * |
| 247 | * Enables crypto on the $requests HTTP socksts and sends the request body asynchronously. |
| 248 | * |
| 249 | * @param Request[] $requests An array of HTTP requests. |
| 250 | */ |
| 251 | protected function enable_crypto( array $requests ) { |
| 252 | foreach ( $this->stream_select( $requests, static::STREAM_SELECT_WRITE ) as $request ) { |
| 253 | // phpcs:ignore WordPress.PHP.NoSilencedErrors.Discouraged |
| 254 | @stream_set_timeout( $this->state->connections[ $request->id ]->http_socket, 1 ); |
| 255 | |
| 256 | // Use @ to suppress warnings. They're collected by error_get_last(). |
| 257 | // phpcs:ignore WordPress.PHP.NoSilencedErrors.Discouraged |
| 258 | $enabled_crypto = @stream_socket_enable_crypto( |
| 259 | $this->state->connections[ $request->id ]->http_socket, |
| 260 | true, |
| 261 | STREAM_CRYPTO_METHOD_TLS_CLIENT |
| 262 | ); |
| 263 | if ( false === $enabled_crypto ) { |
| 264 | $last_error = error_get_last(); |
| 265 | $this->set_error( |
| 266 | $request, |
| 267 | new HttpError( 'Failed to enable crypto: ' . ( is_array( $last_error ) ? $last_error['message'] : 'unknown' ) ) |
| 268 | ); |
| 269 | continue; |
| 270 | } elseif ( 0 === $enabled_crypto ) { |
| 271 | // The SSL handshake isn't finished yet, let's skip it |
| 272 | // for now and try again on the next event loop pass. |
| 273 | continue; |
| 274 | } |
| 275 | // SSL connection established, let's send the headers. |
| 276 | $request->state = Request::STATE_WILL_SEND_HEADERS; |
| 277 | } |
| 278 | } |
| 279 | |
| 280 | /** |
| 281 | * Sends HTTP request headers. |
| 282 | * |
| 283 | * @param Request[] $requests An array of HTTP requests. |
| 284 | */ |
| 285 | protected function send_request_headers( array $requests ) { |
| 286 | foreach ( $this->stream_select( $requests, static::STREAM_SELECT_WRITE ) as $request ) { |
| 287 | $header_bytes = static::prepare_request_headers( $request ); |
| 288 | // phpcs:ignore WordPress.PHP.NoSilencedErrors.Discouraged |
| 289 | if ( false === @fwrite( $this->state->connections[ $request->id ]->http_socket, $header_bytes ) ) { |
| 290 | $last_error = error_get_last(); |
| 291 | $last_error_message = is_array( $last_error ) ? $last_error['message'] : 'unknown'; |
| 292 | $this->set_error( |
| 293 | $request, |
| 294 | new HttpError( 'Failed to write request bytes - ' . $last_error_message ) |
| 295 | ); |
| 296 | continue; |
| 297 | } |
| 298 | |
| 299 | if ( $request->upload_body_stream ) { |
| 300 | $request->state = Request::STATE_WILL_SEND_BODY; |
| 301 | |
| 302 | if ( 'chunked' === $request->get_header( 'transfer-encoding' ) ) { |
| 303 | $request->upload_body_stream = new TransformedReadStream( |
| 304 | $request->upload_body_stream, |
| 305 | array( new ChunkedEncoderByteTransformer() ) |
| 306 | ); |
| 307 | } |
| 308 | } else { |
| 309 | $request->state = Request::STATE_RECEIVING_HEADERS; |
| 310 | } |
| 311 | } |
| 312 | } |
| 313 | |
| 314 | /** |
| 315 | * Sends HTTP request body. |
| 316 | * |
| 317 | * @param Request[] $requests An array of HTTP requests. |
| 318 | */ |
| 319 | protected function send_request_body( array $requests ) { |
| 320 | foreach ( $this->stream_select( $requests, self::STREAM_SELECT_WRITE ) as $request ) { |
| 321 | if ( $request->upload_body_stream->reached_end_of_data() ) { |
| 322 | $request->upload_body_stream->close_reading(); |
| 323 | $request->upload_body_stream = null; |
| 324 | $request->state = Request::STATE_RECEIVING_HEADERS; |
| 325 | continue; |
| 326 | } |
| 327 | |
| 328 | $available_bytes = $request->upload_body_stream->pull( 65536 ); |
| 329 | if ( 0 === $available_bytes ) { |
| 330 | // Not all pull() calls must yield bytes, maybe we just need to wait for the next chunk. |
| 331 | // Let's continue and keep trying. |
| 332 | // @TODO: Implement a generic timeout mechanism for pull() calls. |
| 333 | continue; |
| 334 | } |
| 335 | |
| 336 | $chunk = $request->upload_body_stream->consume( $available_bytes ); |
| 337 | // phpcs:ignore WordPress.PHP.NoSilencedErrors.Discouraged |
| 338 | if ( ! @fwrite( $this->state->connections[ $request->id ]->http_socket, $chunk ) ) { |
| 339 | $last_error = error_get_last(); |
| 340 | $last_error_message = is_array( $last_error ) ? $last_error['message'] : 'unknown'; |
| 341 | $this->set_error( $request, new HttpError( 'Failed to write request bytes: ' . $last_error_message ) ); |
| 342 | continue; |
| 343 | } |
| 344 | } |
| 345 | } |
| 346 | |
| 347 | /** |
| 348 | * Reads the next received portion of HTTP response headers for multiple requests. |
| 349 | * |
| 350 | * @param array $requests An array of requests. |
| 351 | */ |
| 352 | protected function receive_response_headers( $requests ) { |
| 353 | $nb_headers_received = 0; |
| 354 | |
| 355 | foreach ( $this->stream_select( $requests, static::STREAM_SELECT_READ ) as $request ) { |
| 356 | if ( ! $request->response ) { |
| 357 | $request->response = new Response( $request ); |
| 358 | } |
| 359 | $connection = $this->state->connections[ $request->id ]; |
| 360 | $response = $request->response; |
| 361 | |
| 362 | while ( true ) { |
| 363 | // @TODO: Use a larger chunk size here and then scan for \r\n\r\n. |
| 364 | // 1 seems slow and overly conservative. |
| 365 | if ( |
| 366 | ! $this->state->connections[ $request->id ]->http_socket || |
| 367 | ! is_resource( $this->state->connections[ $request->id ]->http_socket ) || |
| 368 | // phpcs:ignore WordPress.PHP.NoSilencedErrors.Discouraged |
| 369 | @feof( $this->state->connections[ $request->id ]->http_socket ) |
| 370 | ) { |
| 371 | $this->set_error( $request, new HttpError( 'Connection closed while reading response headers.' ) ); |
| 372 | break; |
| 373 | } |
| 374 | |
| 375 | $header_byte = fread( $this->state->connections[ $request->id ]->http_socket, 1 ); |
| 376 | |
| 377 | if ( false === $header_byte || '' === $header_byte ) { |
| 378 | if ( |
| 379 | ! $this->state->connections[ $request->id ]->http_socket || |
| 380 | ! is_resource( $this->state->connections[ $request->id ]->http_socket ) || |
| 381 | // phpcs:ignore WordPress.PHP.NoSilencedErrors.Discouraged |
| 382 | @feof( $this->state->connections[ $request->id ]->http_socket ) |
| 383 | ) { |
| 384 | $this->set_error( $request, new HttpError( 'Connection closed while reading response headers.' ) ); |
| 385 | break; |
| 386 | } |
| 387 | break; |
| 388 | } |
| 389 | $connection->response_buffer .= $header_byte; |
| 390 | |
| 391 | $buffer_size = strlen( $connection->response_buffer ); |
| 392 | if ( |
| 393 | $buffer_size < 4 || |
| 394 | "\r" !== $connection->response_buffer[ $buffer_size - 4 ] || |
| 395 | "\n" !== $connection->response_buffer[ $buffer_size - 3 ] || |
| 396 | "\r" !== $connection->response_buffer[ $buffer_size - 2 ] || |
| 397 | "\n" !== $connection->response_buffer[ $buffer_size - 1 ] |
| 398 | ) { |
| 399 | continue; |
| 400 | } |
| 401 | |
| 402 | $request->response = Response::from_http_headers( |
| 403 | $connection->response_buffer, |
| 404 | $request |
| 405 | ); |
| 406 | $connection->response_buffer = ''; |
| 407 | if ( false === $request->response ) { |
| 408 | $this->set_error( $request, new HttpError( 'Malformed HTTP headers received from the server.' ) ); |
| 409 | break; |
| 410 | } |
| 411 | |
| 412 | $this->state->events[ $request->id ][ Client::EVENT_GOT_HEADERS ] = true; |
| 413 | ++$nb_headers_received; |
| 414 | |
| 415 | if ( 0 === $response->total_bytes ) { |
| 416 | $request->state = Request::STATE_RECEIVED; |
| 417 | break; |
| 418 | } |
| 419 | |
| 420 | $request->state = Request::STATE_RECEIVING_BODY; |
| 421 | $this->state->connections[ $request->id ]->decoded_response_stream = $this->decode_and_monitor_response_body_stream( $request ); |
| 422 | break; |
| 423 | } |
| 424 | } |
| 425 | |
| 426 | return $nb_headers_received; |
| 427 | } |
| 428 | |
| 429 | /** |
| 430 | * Reads the next received portion of HTTP response headers for multiple requests. |
| 431 | * |
| 432 | * @param array $requests An array of requests. |
| 433 | */ |
| 434 | protected function receive_response_body( $requests ) { |
| 435 | // @TODO: Assume body is fully received when either. |
| 436 | // * Content-Length is reached. |
| 437 | // * The last chunk in Transfer-Encoding: chunked is received. |
| 438 | // * The connection is closed. |
| 439 | foreach ( $this->stream_select( $requests, static::STREAM_SELECT_READ ) as $request ) { |
| 440 | $stream = $this->state->connections[ $request->id ]->decoded_response_stream; |
| 441 | |
| 442 | while ( true ) { |
| 443 | $available_bytes = $stream->pull( 65536 ); |
| 444 | if ( $available_bytes > 0 ) { |
| 445 | $body_chunk = $stream->consume( $available_bytes ); |
| 446 | $request->response->received_bytes += $available_bytes; |
| 447 | $this->state->connections[ $request->id ]->response_buffer .= $body_chunk; |
| 448 | $this->state->events[ $request->id ][ Client::EVENT_BODY_CHUNK_AVAILABLE ] = true; |
| 449 | break; // Process one chunk per loop iteration. |
| 450 | } elseif ( $stream->reached_end_of_data() ) { |
| 451 | $request->state = Request::STATE_RECEIVED; |
| 452 | break; |
| 453 | } |
| 454 | } |
| 455 | } |
| 456 | } |
| 457 | |
| 458 | /** |
| 459 | * Prepares an HTTP request string for a given URL. |
| 460 | * |
| 461 | * @param Request $request The Request to prepare the HTTP headers for. |
| 462 | * |
| 463 | * @return string The prepared HTTP request string. |
| 464 | */ |
| 465 | protected static function prepare_request_headers( Request $request ) { |
| 466 | $url = $request->url; |
| 467 | $parts = parse_url( $url ); |
| 468 | $path = ( isset( $parts['path'] ) ? $parts['path'] : '/' ) . ( isset( $parts['query'] ) ? '?' . $parts['query'] : '' ); |
| 469 | |
| 470 | $headers = $request->headers; |
| 471 | |
| 472 | /** |
| 473 | * Disable the gzip transfer compression when requesting a byte range. |
| 474 | * |
| 475 | * When we're requesting a byte range AND gzipped transfer encoding, |
| 476 | * our intention is to get compressed bytes 0-X of the original file. |
| 477 | * |
| 478 | * However, some servers will compress the file first, and then return |
| 479 | * the compressed bytes 0-X. The result is both unpredictable and impossible |
| 480 | * to decompress. |
| 481 | */ |
| 482 | if ( ! array_key_exists( 'range', $headers ) && ! array_key_exists( 'accept-encoding', $headers ) ) { |
| 483 | $headers['accept-encoding'] = 'gzip'; |
| 484 | } |
| 485 | |
| 486 | $request_parts = array( |
| 487 | "$request->method $path HTTP/$request->http_version", |
| 488 | ); |
| 489 | |
| 490 | foreach ( $headers as $name => $value ) { |
| 491 | $request_parts[] = "$name: $value"; |
| 492 | } |
| 493 | |
| 494 | return implode( "\r\n", $request_parts ) . "\r\n\r\n"; |
| 495 | } |
| 496 | |
| 497 | protected function filter_requests( array $requests, $states ) { |
| 498 | if ( ! is_array( $states ) ) { |
| 499 | $states = array( $states ); |
| 500 | } |
| 501 | $results = array(); |
| 502 | foreach ( $requests as $request ) { |
| 503 | if ( in_array( $request->state, $states, true ) ) { |
| 504 | $results[] = $request; |
| 505 | } |
| 506 | } |
| 507 | |
| 508 | return $results; |
| 509 | } |
| 510 | |
| 511 | |
| 512 | protected function stream_select( $requests, $mode ) { |
| 513 | if ( empty( $requests ) ) { |
| 514 | return array(); |
| 515 | } |
| 516 | |
| 517 | $read = array(); |
| 518 | $write = array(); |
| 519 | foreach ( $requests as $k => $request ) { |
| 520 | if ( $mode & static::STREAM_SELECT_READ ) { |
| 521 | $read[ $k ] = $this->state->connections[ $request->id ]->http_socket; |
| 522 | } |
| 523 | if ( $mode & static::STREAM_SELECT_WRITE ) { |
| 524 | $write[ $k ] = $this->state->connections[ $request->id ]->http_socket; |
| 525 | } |
| 526 | } |
| 527 | $except = null; |
| 528 | if ( 0 === count( $read ) && 0 === count( $write ) ) { |
| 529 | return array(); |
| 530 | } |
| 531 | |
| 532 | // phpcs:disable WordPress.PHP.NoSilencedErrors.Discouraged |
| 533 | $ready = @stream_select( $read, $write, $except, 0, ClientState::NONBLOCKING_TIMEOUT_MICROSECONDS ); |
| 534 | if ( false === $ready ) { |
| 535 | foreach ( $requests as $request ) { |
| 536 | $this->set_error( $request, new HttpError( 'Error: ' . error_get_last()['message'] ) ); |
| 537 | } |
| 538 | |
| 539 | return array(); |
| 540 | } elseif ( $ready <= 0 ) { |
| 541 | // @TODO allow at most X stream_select attempts per request |
| 542 | // foreach ( $unprocessed_requests as $request ) { |
| 543 | // $this->>set_error($request, new HttpError( 'stream_select timed out' )); |
| 544 | // } |
| 545 | return array(); |
| 546 | } |
| 547 | |
| 548 | $selected_requests = array(); |
| 549 | foreach ( array_keys( $read ) as $k ) { |
| 550 | $selected_requests[ $k ] = $requests[ $k ]; |
| 551 | } |
| 552 | foreach ( array_keys( $write ) as $k ) { |
| 553 | $selected_requests[ $k ] = $requests[ $k ]; |
| 554 | } |
| 555 | |
| 556 | return $selected_requests; |
| 557 | } |
| 558 | |
| 559 | private function mark_finished( Request $request ) { |
| 560 | $this->state->set_request_finished( $request ); |
| 561 | $this->close_connection( $request ); |
| 562 | } |
| 563 | |
| 564 | private function set_error( Request $request, $error ) { |
| 565 | $this->state->set_request_error( $request, $error ); |
| 566 | $this->close_connection( $request ); |
| 567 | } |
| 568 | |
| 569 | private function close_connection( Request $request ) { |
| 570 | $socket = $this->state->connections[ $request->id ]->http_socket; |
| 571 | if ( $socket && is_resource( $socket ) ) { |
| 572 | // Close the TCP socket. |
| 573 | if ( $this->state->connections[ $request->id ]->decoded_response_stream ) { |
| 574 | $stream = $this->state->connections[ $request->id ]->decoded_response_stream; |
| 575 | $stream->close_reading(); |
| 576 | $this->state->connections[ $request->id ]->decoded_response_stream = null; |
| 577 | } else { |
| 578 | @fclose( $socket ); |
| 579 | } |
| 580 | } |
| 581 | } |
| 582 | } |
| 583 |