Allow passing a stream as data input

library/Requests.php

@@ -572,6 +572,16 @@ protected static function set_defaults(&$url, &$headers, &$data, &$type, &$optio
 				$options['data_format'] = 'body';
 			}
 		}
+
+		// Check the data stream is valid
+		if (is_resource($data)) {
+			if ($options['data_format'] !== 'body') {
+				throw new Requests_Exception('Streams can only be sent in the request body.', 'requests.stream_as_query', $data);
+			}
+			if (get_resource_type($data) !== 'stream') {
+				throw new Requests_Exception('Invalid stream resource for request body.', 'requests.invalid_stream', $data);
+			}
+		}
 	}
 
 	/**

library/Requests/Transport/cURL.php

@@ -297,29 +297,48 @@ protected function setup_handle($url, $headers, $data, $options) {
 
 			if ($data_format === 'query') {
 				$url = self::format_get($url, $data);
-				$data = '';
 			}
-			elseif (!is_string($data)) {
-				$data = http_build_query($data, null, '&');
+			elseif ($options['type'] !== Requests::TRACE) {
+				if (is_resource($data)) {
+					$stat = fstat($data);
+					if (!$stat) {
+						throw new Requests_Exception('Body stream resource does not support stat.', 'requests.stream_no_stat', $stat);
+					}
+					curl_setopt($this->handle, CURLOPT_INFILE, $data);
+					curl_setopt($this->handle, CURLOPT_INFILESIZE, $stat['size']);
+
+					// We need to set CURLOPT_PUT so that cURL uses INFILE, but
+					// this will set the request type to PUT by default. We need
+					// to set CURLOPT_CUSTOMREQUEST to set it back.
+					curl_setopt($this->handle, CURLOPT_PUT, true);
+					curl_setopt($this->handle, CURLOPT_CUSTOMREQUEST, $options['type']);
+				}
+				elseif (!is_string($data)) {
+					$data = http_build_query($data, null, '&');
+					curl_setopt($this->handle, CURLOPT_POSTFIELDS, $data);
+				}
+				else {
+					curl_setopt($this->handle, CURLOPT_POSTFIELDS, $data);
+				}
 			}
 		}
 
 		switch ($options['type']) {
 			case Requests::POST:
 				curl_setopt($this->handle, CURLOPT_POST, true);
-				curl_setopt($this->handle, CURLOPT_POSTFIELDS, $data);
+				if ($data_format === 'query') {
+					// cURL needs a POST body, even if we didn't have one.
+					curl_setopt($this->handle, CURLOPT_POSTFIELDS, '');
+				}
 				break;
+
+			case Requests::HEAD:
+				curl_setopt($this->handle, CURLOPT_NOBODY, true);
+				// Fall-through
 			case Requests::PATCH:
 			case Requests::PUT:
 			case Requests::DELETE:
 			case Requests::OPTIONS:
-				curl_setopt($this->handle, CURLOPT_CUSTOMREQUEST, $options['type']);
-				curl_setopt($this->handle, CURLOPT_POSTFIELDS, $data);
-				break;
-			case Requests::HEAD:
-				curl_setopt($this->handle, CURLOPT_CUSTOMREQUEST, $options['type']);
-				curl_setopt($this->handle, CURLOPT_NOBODY, true);
-				break;
 			case Requests::TRACE:
 				curl_setopt($this->handle, CURLOPT_CUSTOMREQUEST, $options['type']);
 				break;

library/Requests/Transport/fsockopen.php

@@ -34,6 +34,13 @@ class Requests_Transport_fsockopen implements Requests_Transport {
 	 */
 	public $info;
 
+	/**
+	 * Request body to send.
+	 *
+	 * @var stream|string|null
+	 */
+	protected $request_body = null;
+
 	/**
 	 * What's the maximum number of bytes we should keep?
 	 *
@@ -138,35 +145,19 @@ public function request($url, $headers = array(), $data = array(), $options = ar
 
 		if ($data_format === 'query') {
 			$path = self::format_get($url_parts, $data);
-			$data = '';
+			$data = null;
 		}
 		else {
 			$path = self::format_get($url_parts, array());
 		}
 
 		$options['hooks']->dispatch('fsockopen.remote_host_path', array(&$path, $url));
 
-		$request_body = '';
+		$this->request_body = '';
 		$out = sprintf("%s %s HTTP/%.1f\r\n", $options['type'], $path, $options['protocol_version']);
 
-		if ($options['type'] !== Requests::TRACE) {
-			if (is_array($data)) {
-				$request_body = http_build_query($data, null, '&');
-			}
-			else {
-				$request_body = $data;
-			}
-
-			if (!empty($data)) {
-				if (!isset($case_insensitive_headers['Content-Length'])) {
-					$headers['Content-Length'] = strlen($request_body);
-				}
-
-				if (!isset($case_insensitive_headers['Content-Type'])) {
-					$headers['Content-Type'] = 'application/x-www-form-urlencoded; charset=UTF-8';
-				}
-			}
-		}
+		$body_headers = $this->prepare_body($data, $case_insensitive_headers, $options);
+		$headers = array_merge($headers, $body_headers);
 
 		if (!isset($case_insensitive_headers['Host'])) {
 			$out .= sprintf('Host: %s', $url_parts['host']);
@@ -202,11 +193,16 @@ public function request($url, $headers = array(), $data = array(), $options = ar
 			$out .= "Connection: Close\r\n";
 		}
 
-		$out .= "\r\n" . $request_body;
+		$out .= "\r\n";
+		if (is_string($this->request_body)) {
+			$out .= $this->request_body;
+		}
 
 		$options['hooks']->dispatch('fsockopen.before_send', array(&$out));
 
 		fwrite($socket, $out);
+		$this->send_body($socket);
+
 		$options['hooks']->dispatch('fsockopen.after_send', array($out));
 
 		if (!$options['blocking']) {
@@ -317,6 +313,69 @@ public function request_multiple($requests, $options) {
 		return $responses;
 	}
 
+	/**
+	 * Prepare the body data to send.
+	 *
+	 * @param string|resource|null $data Data as a string, stream resource, or null.
+	 * @param array|Requests_Utility_CaseInsensitiveDictionary $headers Headers set on the request.
+	 * @param array $options Options set on the request.
+	 * @return array Extra headers to add to the request.
+	 */
+	protected function prepare_body($data, $headers, $options) {
+		if (empty($data)) {
+			return array();
+		}
+
+		$body_headers = array();
+		if ($options['type'] !== Requests::TRACE) {
+			if (is_array($data)) {
+				$this->request_body = http_build_query($data, null, '&');
+				$length = strlen($this->request_body);
+			}
+			elseif (is_resource($data)) {
+				$this->request_body = $data;
+				$stat = fstat($data);
+				if (!$stat) {
+					throw new Requests_Exception('Body stream resource does not support stat.', 'requests.stream_no_stat', $stat);
+				}
+				$length = $stat['size'];
+			}
+			else {
+				$this->request_body = $data;
+				$length = strlen($this->request_body);
+			}
+
+			if (!empty($data)) {
+				if (!isset($headers['Content-Length'])) {
+					$body_headers['Content-Length'] = $length;
+				}
+
+				if (!isset($headers['Content-Type'])) {
+					$body_headers['Content-Type'] = 'application/x-www-form-urlencoded; charset=UTF-8';
+				}
+			}
+		}
+
+		return $body_headers;
+	}
+
+	/**
+	 * Send body data with the request.
+	 *
+	 * @param resource $stream Remote socket for the server.
+	 */
+	protected function send_body($stream) {
+		if (!is_resource($this->request_body)) {
+			// Already sent
+			return;
+		}
+
+		while (!feof($this->request_body)) {
+			$bytes = fread($this->request_body, Requests::BUFFER_SIZE);
+			fwrite($stream, $bytes);
+		}
+	}
+
 	/**
 	 * Retrieve the encodings we can accept
 	 *

tests/Transport/Base.php

@@ -181,6 +181,88 @@ public function testPOSTWithNestedData() {
 		$this->assertEquals(array('test' => 'true', 'test2[test3]' => 'test', 'test2[test4]' => 'test-too'), $result['form']);
 	}
 
+	public function streamProvider() {
+		$streams = array();
+
+		// Regular stream
+		$contents = file_get_contents(__FILE__);
+		$stream = fopen(__FILE__, 'r');
+		$streams[] = array($stream, $contents, strlen($contents));
+
+		// In-memory stream
+		$string = "Hello! \xF0\x9F\x92\xA9";
+		$stream = fopen('php://memory', 'r+');
+		fwrite($stream, $string);
+
+		// Reset for reading
+		rewind($stream);
+		$streams[] = array($stream, $string, strlen($string));
+
+		return $streams;
+	}
+
+	/**
+	 * @dataProvider streamProvider
+	 */
+	public function testPOSTWithStream($stream, $expected_data, $expected_length) {
+		$request = Requests::post(httpbin('/post'), array(), $stream, $this->getOptions());
+		fclose($stream);
+		$this->assertEquals(200, $request->status_code);
+
+		$result = json_decode($request->body, true);
+		$this->assertEquals($expected_data, $result['data']);
+
+		// Check the length we sent
+		$sent_headers = new Requests_Utility_CaseInsensitiveDictionary($result['headers']);
+		$this->assertEquals($expected_length, $sent_headers['Content-Length']);
+	}
+
+	public function testPOSTWithInvalidStream() {
+		// Register our wrapper
+		stream_wrapper_register('requeststestvar', 'RequestsTest_VariableStream');
+
+		$data = base64_encode('hello');
+		$stream = fopen('requeststestvar://', 'r+');
+		stream_wrapper_unregister('requeststestvar');
+
+		fwrite($stream, 'Hello!');
+		rewind($stream);
+
+		// This should fail, as the stream doesn't support stat
+		$this->setExpectedException('Requests_Exception', 'Body stream resource does not support stat.');
+
+		$request = Requests::post(httpbin('/post'), array(), $stream, $this->getOptions());
+		fclose($stream);
+	}
+
+	public function testPOSTStreamInQuery() {
+		$stream = fopen('php://memory', 'r');
+		$options = array(
+			// Attempt to use the stream for query data
+			'data_format' => 'query',
+		);
+
+		// This should fail, as streams can't be used for query data
+		$this->setExpectedException('Requests_Exception', 'Streams can only be sent in the request body.');
+
+		$request = Requests::post(httpbin('/post'), array(), $stream, $this->getOptions($options));
+	}
+
+	public function testPOSTWithInvalidResource() {
+		// Use a socket stream instead of a file
+		$stream = socket_create(AF_UNIX, SOCK_STREAM, 0);
+
+		// This should fail, as the resource isn't a stream
+		$this->setExpectedException('Requests_Exception', 'Invalid stream resource for request body.');
+
+		// https://github.com/facebook/hhvm/issues/4036
+		if (defined('HHVM_VERSION')) {
+			$this->setExpectedException('Requests_Exception', 'Body stream resource does not support stat.');
+		}
+
+		$request = Requests::post(httpbin('/post'), array(), $stream, $this->getOptions());
+	}
+
 	public function testRawPUT() {
 		$data = 'test';
 		$request = Requests::put(httpbin('/put'), array(), $data, $this->getOptions());
@@ -211,6 +293,22 @@ public function testPUTWithArray() {
 		$this->assertEquals(array('test' => 'true', 'test2' => 'test'), $result['form']);
 	}
 
+	/**
+	 * @dataProvider streamProvider
+	 */
+	public function testPUTWithStream($stream, $expected_data, $expected_length) {
+		$request = Requests::put(httpbin('/put'), array(), $stream, $this->getOptions());
+		fclose($stream);
+		$this->assertEquals(200, $request->status_code);
+
+		$result = json_decode($request->body, true);
+		$this->assertEquals($expected_data, $result['data']);
+
+		// Check the length we sent
+		$sent_headers = new Requests_Utility_CaseInsensitiveDictionary($result['headers']);
+		$this->assertEquals($expected_length, $sent_headers['Content-Length']);
+	}
+
 	public function testRawPATCH() {
 		$data = 'test';
 		$request = Requests::patch(httpbin('/patch'), array(), $data, $this->getOptions());