Merge pull request #9 from php-api-clients/feature-streaming-body

Streaming body

Author: WyriHaximus

src/Transport/Client.php

@@ -16,6 +16,7 @@
 use Psr\Http\Message\UriInterface;
 use React\Cache\CacheInterface;
 use React\EventLoop\LoopInterface;
+use React\EventLoop\Timer\TimerInterface;
 use React\Promise\PromiseInterface;
 use function React\Promise\reject;
 use React\Promise\RejectedPromise;
@@ -209,12 +210,9 @@ public function request(RequestInterface $request, array $options = [], bool $re
             ));
         })->then(function (ResponseInterface $response) use ($request, $options, $refresh) {
             if (isset($options[RequestOptions::STREAM]) && $options[RequestOptions::STREAM] === true) {
-                return resolve(
-                    new Response(
-                        '',
-                        $response
-                    )
-                );
+                $responseWrapper = new Response('', $response);
+                $this->streamBody($responseWrapper);
+                return resolve($responseWrapper);
             }
 
             $contents = $response->getBody()->getContents();
@@ -245,6 +243,25 @@ public function request(RequestInterface $request, array $options = [], bool $re
         });
     }
 
+    protected function streamBody(Response $response)
+    {
+        $stream = $response->getResponse()->getBody();
+        $this->loop->addPeriodicTimer(0.001, function (TimerInterface $timer) use ($stream, $response) {
+            if ($stream->eof()) {
+                $timer->cancel();
+                $response->emit('end');
+                return;
+            }
+
+            $size = $stream->getSize();
+            if ($size === 0) {
+                return;
+            }
+
+            $response->emit('data', [$stream->read($size)]);
+        });
+    }
+
     protected function applyApiSettingsToRequest(RequestInterface $request): RequestInterface
     {
         $uri = $request->getUri();

src/Transport/Response.php

@@ -2,10 +2,14 @@
 
 namespace ApiClients\Foundation\Transport;
 
+use Evenement\EventEmitterInterface;
+use Evenement\EventEmitterTrait;
 use Psr\Http\Message\ResponseInterface;
 
-final class Response
+final class Response implements EventEmitterInterface
 {
+    use EventEmitterTrait;
+
     /**
      * @var string
      */

tests/Transport/ClientTest.php

@@ -241,7 +241,22 @@ public function testRequestStreaming()
         $cache = Phake::mock(CacheInterface::class);
 
         $stream = Phake::mock(StreamInterface::class);
-        Phake::when($stream)->getContents()->thenReturn('{"foo":"bar"}');
+        Phake::when($stream)->eof()
+            ->thenReturn(false)
+            ->thenReturn(false)
+            ->thenReturn(false)
+            ->thenReturn(true)
+        ;
+        Phake::when($stream)->getSize()
+            ->thenReturn(1)
+            ->thenReturn(1)
+            ->thenReturn(1)
+        ;
+        Phake::when($stream)->read(1)
+            ->thenReturn('a')
+            ->thenReturn('b')
+            ->thenReturn('c')
+        ;
 
         $response = Phake::mock(Response::class);
         Phake::when($response)->getBody()->thenReturn($stream);
@@ -276,6 +291,14 @@ public function testRequestStreaming()
         $this->assertInstanceOf(TransportResponse::class, $result);
         $this->assertSame('', $result->getBody());
 
+        $buffer = '';
+        $result->on('data', function ($data) use (&$buffer) {
+            $buffer .= $data;
+        });
+        $loop->run();
+
+        $this->assertSame('abc', $buffer);
+
         Phake::verifyNoInteraction($cache);
     }