[HttpClient] fix support for non-blocking resource streams

src/Symfony/Component/HttpClient/Response/StreamWrapper.php

@@ -37,6 +37,8 @@ class StreamWrapper
     /** @var resource|null */
     private $handle;
 
+    private $blocking = true;
+    private $timeout;
     private $eof = false;
     private $offset = 0;
 
@@ -147,7 +149,7 @@ class StreamWrapper
             return $data;
         }
 
-        foreach ($this->client->stream([$this->response]) as $chunk) {
+        foreach ($this->client->stream([$this->response], $this->blocking ? $this->timeout : 0) as $chunk) {
             try {
                 $this->eof = true;
                 $this->eof = !$chunk->isTimeout();
@@ -178,6 +180,19 @@ class StreamWrapper
         return '';
     }
 
+    public function stream_set_option(int $option, int $arg1, ?int $arg2): bool
+    {
+        if (STREAM_OPTION_BLOCKING === $option) {
+            $this->blocking = (bool) $arg1;
+        } elseif (STREAM_OPTION_READ_TIMEOUT === $option) {
+            $this->timeout = $arg1 + $arg2 / 1e6;
+        } else {
+            return false;
+        }
+
+        return true;
+    }
+
     public function stream_tell(): int
     {
         return $this->offset;

src/Symfony/Component/HttpClient/Tests/HttpClientTestCase.php

@@ -75,4 +75,20 @@ abstract class HttpClientTestCase extends BaseHttpClientTestCase
         $response = $client->request('GET', 'http://localhost:8057/404');
         $stream = $response->toStream();
     }
+
+    public function testNonBlockingStream()
+    {
+        $client = $this->getHttpClient(__FUNCTION__);
+        $response = $client->request('GET', 'http://localhost:8057/timeout-body');
+        $stream = $response->toStream();
+
+        $this->assertTrue(stream_set_blocking($stream, false));
+        $this->assertSame('<1>', fread($stream, 8192));
+        $this->assertFalse(feof($stream));
+
+        $this->assertTrue(stream_set_blocking($stream, true));
+        $this->assertSame('<2>', fread($stream, 8192));
+        $this->assertSame('', fread($stream, 8192));
+        $this->assertTrue(feof($stream));
+    }
 }

src/Symfony/Component/HttpClient/Tests/MockHttpClientTest.php

@@ -171,6 +171,10 @@ class MockHttpClientTest extends HttpClientTestCase
 
                 return $client;
 
+            case 'testNonBlockingStream':
+                $responses[] = new MockResponse((function () { yield '<1>'; yield ''; yield '<2>'; })(), ['response_headers' => $headers]);
+                break;
+
             case 'testMaxDuration':
                 $mock = $this->getMockBuilder(ResponseInterface::class)->getMock();
                 $mock->expects($this->any())