Add StreamingParser::parsePlainStream() helper

Author: clue

src/Io/StreamingParser.php

@@ -14,9 +14,44 @@ class StreamingParser
 {
     public function parseJsonStream(PromiseInterface $promise)
     {
-        // TODO: assert expect tcp stream
+        // application/json
 
+        $in = $this->parsePlainStream($promise);
+        $out = new ReadableStream();
+
+        // invalid/closing input stream => return closed output stream
+        if (!$in->isReadable()) {
+            $out->close();
+
+            return $out;
+        }
+
+        // forward each data chunk to the streaming JSON parser
         $parser = new StreamingJsonParser();
+        $in->on('data', function ($data) use ($parser, $out) {
+            $objects = $parser->push($data);
+
+            foreach ($objects as $object) {
+                $out->emit('progress', array($object, $out));
+            }
+        });
+
+        // forward error and make sure stream closes
+        $in->on('error', function ($error) use ($out) {
+            $out->emit('error', array($error, $out));
+            $out->close();
+        });
+
+        // closing either stream closes the other one
+        $in->on('close', array($out, 'close'));
+        $out->on('close', array($in, 'close'));
+
+        return $out;
+    }
+
+    public function parsePlainStream(PromiseInterface $promise)
+    {
+        // text/plain
 
         $out = new ReadableStream();
 
@@ -35,21 +70,17 @@ function ($error) use ($out) {
                 $out->emit('error', array($error, $out));
                 $out->close();
             },
-            function ($progress) use ($parser, $out) {
+            function ($progress) use ($out) {
                 if (is_array($progress) && isset($progress['responseStream'])) {
                     $stream = $progress['responseStream'];
                     /* @var $stream React\Stream\Stream */
 
                     // hack to do not buffer stream contents in body
                     $stream->removeAllListeners('data');
 
-                    // got a streaming HTTP reponse => forward each data chunk to the streaming JSON parser
-                    $stream->on('data', function ($data) use ($parser, $out) {
-                        $objects = $parser->push($data);
-
-                        foreach ($objects as $object) {
-                            $out->emit('progress', array($object, $out));
-                        }
+                    // got a streaming HTTP response => forward each data chunk to the resulting output stream
+                    $stream->on('data', function ($data) use ($out) {
+                        $out->emit('data', array($data, $out));
                     });
                 }
             }

tests/Io/StreamingParserTest.php

@@ -0,0 +1,123 @@
+<?php
+
+use Clue\React\Docker\Io\StreamingParser;
+use React\Promise\Deferred;
+use React\Stream\ReadableStream;
+use React\Promise\CancellablePromiseInterface;
+
+class StreamingParserTest extends TestCase
+{
+    private $parser;
+
+    public function setUp()
+    {
+        $this->parser = new StreamingParser();
+    }
+
+    public function testJsonPassingRejectedPromiseResolvesWithClosedStream()
+    {
+        $stream = $this->parser->parseJsonStream($this->createPromiseRejected());
+
+        $this->assertInstanceOf('React\Stream\ReadableStreamInterface', $stream);
+        $this->assertFalse($stream->isReadable());
+    }
+
+    public function testJsonRejectingPromiseWillEmitErrorAndCloseEvent()
+    {
+        $deferred = new Deferred();
+
+        $stream = $this->parser->parseJsonStream($deferred->promise());
+
+        $this->assertTrue($stream->isReadable());
+
+        $exception = new RuntimeException();
+
+        $stream->on('error', $this->expectCallableOnceWith($exception));
+        $stream->on('close', $this->expectCallableOnce());
+
+        $deferred->reject($exception);
+
+        $this->assertFalse($stream->isReadable());
+    }
+
+    public function testJsonResolvingPromiseWillEmitCloseEvent()
+    {
+        $deferred = new Deferred();
+
+        $stream = $this->parser->parseJsonStream($deferred->promise());
+
+        $this->assertTrue($stream->isReadable());
+
+        $stream->on('error', $this->expectCallableNever());
+        $stream->on('close', $this->expectCallableOnce());
+
+        $deferred->resolve('data');
+
+        $this->assertFalse($stream->isReadable());
+    }
+
+    public function testPlainPassingRejectedPromiseResolvesWithClosedStream()
+    {
+        $stream = $this->parser->parsePlainStream($this->createPromiseRejected());
+
+        $this->assertInstanceOf('React\Stream\ReadableStreamInterface', $stream);
+        $this->assertFalse($stream->isReadable());
+    }
+
+    public function testDeferredClosedStreamWillReject()
+    {
+        $stream = $this->getMock('React\Stream\ReadableStreamInterface');
+        $stream->expects($this->once())->method('isReadable')->will($this->returnValue(false));
+
+        $promise = $this->parser->deferredStream($stream, 'anything');
+        $this->expectPromiseReject($promise);
+    }
+
+    public function testDeferredStreamEventsWillBeEmittedAndBuffered()
+    {
+        $stream = new ReadableStream();
+
+        $promise = $this->parser->deferredStream($stream, 'demo');
+
+        $stream->emit('ignored', array('ignored'));
+        $stream->emit('demo', array('a'));
+        $stream->emit('demo', array('b'));
+
+        $stream->close();
+
+        $this->expectPromiseResolveWith(array('a', 'b'), $promise);
+    }
+
+    public function testDeferredStreamErrorEventWillRejectPromise()
+    {
+        $stream = new ReadableStream();
+
+        $promise = $this->parser->deferredStream($stream, 'demo');
+
+        $stream->emit('ignored', array('ignored'));
+
+        $stream->emit('demo', array('a'));
+
+        $stream->emit('error', array('value', 'ignord'));
+
+        $stream->close();
+
+        $this->expectPromiseReject($promise);
+        $promise->then(null, $this->expectCallableOnceWith('value'));
+    }
+
+    public function testDeferredCancelingPromiseWillCloseStream()
+    {
+        $this->markTestIncomplete();
+
+        $stream = $this->getMock('React\Stream\ReadableStreamInterface');
+
+        $promise = $this->parser->deferredStream($stream, 'anything');
+        if (!($promise instanceof CancellablePromiseInterface)) {
+            $this->markTestSkipped('Requires Promise v2 API and has no effect on v1 API');
+        }
+
+        $stream->expects($this->once())->method('close');
+        $promise->cancel();
+    }
+}