Add streaming support

docs/hooks.md

@@ -25,6 +25,10 @@ Available Hooks
 
     Parameters: `string $data`, `int $response_bytes`, `int|bool $response_byte_limit`
 
+    When the [`stream`](usage-advanced.md#streaming-responses) option is enabled,
+    this hook fires once per `WpOrg\Requests\Response::read()` call as the caller
+    pulls the body, not while the transport receives the response.
+
 * **`requests.before_parse`**
 
     Alter the raw HTTP response before parsing.
@@ -95,6 +99,11 @@ Available Hooks
 
     Prior to Requests 2.1.0, the `$info` parameter was omitted for non-blocking requests.
 
+    When the [`stream`](usage-advanced.md#streaming-responses) option is enabled,
+    this hook fires once the response headers are in, so the `$info` array
+    reflects the transfer at that point (timing and size fields are not final
+    yet).
+
 * **`curl.before_multi_add`**
 
     Set cURL options before adding the request to a cURL multi handle.

docs/usage-advanced.md

@@ -30,6 +30,67 @@ $session->useragent = 'My-Awesome-App';
 $session->options['useragent'] = 'My-Awesome-App';
 ```
 
+## Streaming Responses
+
+By default, Requests buffers the whole response body in memory before returning.
+For large downloads or long-lived responses, such as Server-Sent Events from an
+AI API, set the `stream` option to `true` instead. The request then returns as
+soon as the response headers are in, and you read the body yourself:
+
+```php
+$response = \WpOrg\Requests\Requests::get('https://example.org/events', array(), array('stream' => true));
+
+// The status code and headers are available right away.
+var_dump($response->status_code);
+
+// Pull the body off the wire in chunks.
+while (!$response->eof()) {
+  $chunk = $response->read();
+  // do something with $chunk
+}
+
+// Release the connection when done.
+$response->close();
+```
+
+The streaming API on `WpOrg\Requests\Response` is small:
+
+- `read($length = 8192)` reads up to `$length` bytes of the body, blocking until
+  data arrives. An empty string means the body has ended.
+- `eof()` tells you whether the body has been fully read.
+- `close()` releases the underlying connection. Also safe to call on responses
+  that are not streamed.
+- `is_streaming()` tells you whether the response is being streamed. Custom
+  transports without streaming support fall back to a normal buffered response,
+  which this lets you detect.
+
+For line-based protocols such as NDJSON or Server-Sent Events, buffer the chunks
+until a newline shows up. See
+[`examples/stream.php`](https://github.com/WordPress/Requests/blob/develop/examples/stream.php)
+for a runnable example.
+
+A few things change when streaming:
+
+- `$response->body` stays empty and `$response->raw` only contains the headers.
+  The body comes out of `read()` instead.
+- Bytes from `read()` are already de-chunked. Both transports ask the server for
+  an uncompressed (`identity`) body, and Requests never decompresses streamed
+  bytes, so what the server sends is what you get.
+- The `timeout` option turns into an idle timeout: the maximum wait for the next
+  chunk of data. A long-lived stream stays open as long as data keeps arriving.
+- `max_bytes` and redirects work as usual. Redirect responses along the way are
+  discarded; only the final response body is streamed.
+- The `request.progress` hook fires once per `read()` call.
+- `stream` needs a blocking request, and cannot be combined with `filename` or
+  with `WpOrg\Requests\Requests::request_multiple()`.
+
+Requests does not implement PSR-7, but `read()`, `eof()` and `close()` match the
+semantics of the corresponding PSR-7 `StreamInterface` methods, so a [PSR-18
+adapter][psr18-adapter] can wrap a streamed body as a readable PSR-7 stream
+without translation.
+
+[psr18-adapter]: https://github.com/Art4/requests-psr18-adapter
+
 
 Secure Requests with SSL
 ------------------------

examples/stream.php

@@ -0,0 +1,45 @@
+<?php
+/**
+ * Requests for PHP, an HTTP library.
+ *
+ * @package   Requests\Examples
+ * @copyright 2012-2023 Requests Contributors
+ * @license   https://github.com/WordPress/Requests/blob/stable/LICENSE ISC
+ * @link      https://github.com/WordPress/Requests
+ */
+
+require_once dirname(__DIR__) . '/src/Autoload.php';
+
+WpOrg\Requests\Autoload::register();
+
+$response = WpOrg\Requests\Requests::get(
+	'https://httpbin.dev/stream/100',
+	['Accept' => 'application/json'],
+	['stream' => true] // Enables streaming mode.
+);
+
+echo 'Status: ', $response->status_code, PHP_EOL;
+echo 'Content-Type: ', $response->headers['content-type'], PHP_EOL, PHP_EOL;
+
+$buffer = '';
+$lines  = 0;
+
+while (!$response->eof()) {
+	$buffer .= $response->read(8192);
+
+	while (($pos = strpos($buffer, "\n")) !== false) {
+		$line   = substr($buffer, 0, $pos);
+		$buffer = substr($buffer, $pos + 1);
+
+		if (trim($line) === '') {
+			continue;
+		}
+
+		var_dump(json_decode($line, true));
+		++$lines;
+	}
+}
+
+echo PHP_EOL, 'Processed ', $lines, ' streamed lines.', PHP_EOL;
+
+$response->close();

src/Requests.php

@@ -125,6 +125,7 @@ class Requests {
 		'blocking'         => true,
 		'type'             => self::GET,
 		'filename'         => false,
+		'stream'           => false,
 		'auth'             => false,
 		'proxy'            => false,
 		'cookies'          => false,
@@ -392,6 +393,14 @@ public static function patch($url, $headers, $data = [], $options = []) {
 	 *    (bool, default: true)
 	 * - `filename`: File to stream the body to instead.
 	 *    (string|bool, default: false)
+	 * - `stream`: Return once the response headers have been received and read
+	 *    the body incrementally via {@see \WpOrg\Requests\Response::read()}
+	 *    instead of buffering it into {@see \WpOrg\Requests\Response::$body}.
+	 *    Cannot be combined with `filename`, requires `blocking` and is not
+	 *    supported for multiple requests. In streaming mode, the `timeout`
+	 *    option acts as an idle timeout (the maximum wait for the next chunk of
+	 *    data) rather than as a total-transfer limit.
+	 *    (bool, default: false)
 	 * - `auth`: Authentication handler or array of user/password details to use
 	 *    for Basic authentication
 	 *    (\WpOrg\Requests\Auth|array|bool, default: false)
@@ -467,9 +476,14 @@ public static function request($url, $headers = [], $data = [], $type = self::GE
 
 		$response = $transport->request($url, $headers, $data, $options);
 
+		$stream = null;
+		if (!empty($options['stream']) && isset($transport->response_stream)) {
+			$stream = $transport->response_stream;
+		}
+
 		$options['hooks']->dispatch('requests.before_parse', [&$response, $url, $headers, $data, $type, $options]);
 
-		return self::parse_response($response, $url, $headers, $data, $options);
+		return self::parse_response($response, $url, $headers, $data, $options, $stream);
 	}
 
 	/**
@@ -515,6 +529,7 @@ public static function request($url, $headers = [], $data = [], $type = self::GE
 	 *
 	 * @throws \WpOrg\Requests\Exception\InvalidArgument When the passed $requests argument is not an array or iterable object with array access.
 	 * @throws \WpOrg\Requests\Exception\InvalidArgument When the passed $options argument is not an array.
+	 * @throws \WpOrg\Requests\Exception When the `stream` option is enabled for any of the requests.
 	 */
 	public static function request_multiple($requests, $options = []) {
 		if (InputValidator::has_array_access($requests) === false || InputValidator::is_iterable($requests) === false) {
@@ -558,6 +573,10 @@ public static function request_multiple($requests, $options = []) {
 				$request['options'] = array_merge($options, $request['options']);
 			}
 
+			if (!empty($request['options']['stream'])) {
+				throw new Exception('The stream option is not supported for multiple requests', 'requests.stream.unsupported_multiple');
+			}
+
 			self::set_defaults($request['url'], $request['headers'], $request['data'], $request['type'], $request['options']);
 
 			// Ensure we only hook in once
@@ -651,12 +670,24 @@ public static function set_certificate_path($path) {
 	 * @return void
 	 *
 	 * @throws \WpOrg\Requests\Exception When the $url is not an http(s) URL.
+	 * @throws \WpOrg\Requests\Exception When the `stream` option is enabled for a non-blocking request.
+	 * @throws \WpOrg\Requests\Exception When the `stream` option is combined with the `filename` option.
 	 */
 	protected static function set_defaults(&$url, &$headers, &$data, &$type, &$options) {
 		if (!preg_match('/^http(s)?:\/\//i', $url, $matches)) {
 			throw new Exception('Only HTTP(S) requests are handled.', 'nonhttp', $url);
 		}
 
+		if (!empty($options['stream'])) {
+			if (empty($options['blocking'])) {
+				throw new Exception('The stream option requires a blocking request', 'requests.stream.requires_blocking');
+			}
+
+			if ($options['filename'] !== false) {
+				throw new Exception('The stream and filename options cannot both be enabled for the same request', 'requests.stream.filename_conflict');
+			}
+		}
+
 		if (empty($options['hooks'])) {
 			$options['hooks'] = new Hooks();
 		}
@@ -713,13 +744,14 @@ protected static function set_defaults(&$url, &$headers, &$data, &$type, &$optio
 	 * @param array  $req_headers Original $headers array passed to {@link request()}, in case we need to follow redirects
 	 * @param array  $req_data    Original $data array passed to {@link request()}, in case we need to follow redirects
 	 * @param array  $options     Original $options array passed to {@link request()}, in case we need to follow redirects
+	 * @param \WpOrg\Requests\Response\Stream|null $stream Live response body stream when the `stream` option is enabled, null otherwise
 	 * @return \WpOrg\Requests\Response
 	 *
 	 * @throws \WpOrg\Requests\Exception On missing head/body separator (`requests.no_crlf_separator`)
 	 * @throws \WpOrg\Requests\Exception On missing head/body separator (`noversion`)
 	 * @throws \WpOrg\Requests\Exception On missing head/body separator (`toomanyredirects`)
 	 */
-	protected static function parse_response($headers, $url, $req_headers, $req_data, $options) {
+	protected static function parse_response($headers, $url, $req_headers, $req_data, $options, $stream = null) {
 		$return = new Response();
 		if (!$options['blocking']) {
 			return $return;
@@ -789,6 +821,12 @@ protected static function parse_response($headers, $url, $req_headers, $req_data
 					$options['type'] = self::GET;
 				}
 
+				// Close the stream if it was opened, as we are about to redirect and will not be using it anymore.
+				if ($stream !== null) {
+					$stream->close();
+					$stream = null;
+				}
+
 				++$options['redirected'];
 				$location = $return->headers['location'];
 				if (strpos($location, 'http://') !== 0 && strpos($location, 'https://') !== 0) {
@@ -815,6 +853,10 @@ protected static function parse_response($headers, $url, $req_headers, $req_data
 
 		$return->redirects = $options['redirected'];
 
+		if ($stream !== null) {
+			$return->set_stream($stream);
+		}
+
 		$options['hooks']->dispatch('requests.after_request', [&$return, $req_headers, $req_data, $options]);
 		return $return;
 	}

src/Response.php

@@ -13,6 +13,7 @@
 use WpOrg\Requests\Exception;
 use WpOrg\Requests\Exception\Http;
 use WpOrg\Requests\Response\Headers;
+use WpOrg\Requests\Response\Stream;
 
 /**
  * HTTP response class
@@ -93,6 +94,15 @@ class Response {
 	 */
 	public $cookies = [];
 
+	/**
+	 * Response body stream.
+	 *
+	 * @since 2.1.0
+	 *
+	 * @var \WpOrg\Requests\Response\Stream|null
+	 */
+	protected $stream = null;
+
 	/**
 	 * Constructor
 	 */
@@ -164,4 +174,73 @@ public function decode_body($associative = true, $depth = 512, $options = 0) {
 
 		return $data;
 	}
+
+	/**
+	 * Set the response body stream.
+	 *
+	 * @since 2.1.0
+	 *
+	 * @param \WpOrg\Requests\Response\Stream $stream Response body stream.
+	 * @return void
+	 */
+	public function set_stream($stream) {
+		$this->stream = $stream;
+	}
+
+	/**
+	 * Check if the response body is being streamed.
+	 *
+	 * @since 2.1.0
+	 *
+	 * @return bool True when the request was made with the `stream` option.
+	 */
+	public function is_streaming() {
+		return $this->stream instanceof Stream;
+	}
+
+	/**
+	 * Read from the streamed response body.
+	 *
+	 * @param int $length Optional. Maximum number of bytes to read.
+	 * @return string Body bytes, already de-chunked. An empty string indicates
+	 *                the end of the body.
+	 *
+	 * @throws \WpOrg\Requests\Exception When the response is not being streamed (`response.not_streaming`).
+	 * @throws \WpOrg\Requests\Exception\InvalidArgument When the passed $length argument is not a positive integer.
+	 */
+	public function read($length = Stream::DEFAULT_CHUNK_SIZE) {
+		if ($this->is_streaming() === false) {
+			throw new Exception('Response is not being streamed', 'response.not_streaming', $this);
+		}
+
+		return $this->stream->read($length);
+	}
+
+	/**
+	 * Check if the end of the streamed response body has been reached.
+	 *
+	 * @since 2.1.0
+	 *
+	 * @return bool
+	 *
+	 * @throws \WpOrg\Requests\Exception When the response is not being streamed.
+	 */
+	public function eof() {
+		if ($this->is_streaming() === false) {
+			throw new Exception('Response is not being streamed', 'response.not_streaming', $this);
+		}
+
+		return $this->stream->eof();
+	}
+
+	/**
+	 * Close the streamed response body.
+	 *
+	 * @return void
+	 */
+	public function close() {
+		if ($this->stream instanceof Stream) {
+			$this->stream->close();
+		}
+	}
 }