Simplify event forwarding semantics to not generate additional events

Author: clue

src/UnwrapReadableStream.php

@@ -47,19 +47,23 @@ function (ReadableStreamInterface $stream) use ($out) {
                     $out->emit('data', array($data, $out));
                 });
 
+                // forward end events and close
+                $stream->on('end', function () use ($out) {
+                    $out->emit('end', array($out));
+                    $out->close();
+                });
+
                 // error events cancel output stream
                 $stream->on('error', function ($error) use ($out) {
                     $out->emit('error', array($error, $out));
                     $out->close();
                 });
 
-                // close output stream once body closes
+                // close output stream once input closes
                 $stream->on('close', function () use ($out) {
                     $out->close();
                 });
-                $stream->on('end', function () use ($out) {
-                    $out->close();
-                });
+
                 return $stream;
             },
             function ($e) use ($out) {
@@ -108,7 +112,6 @@ public function close()
             $this->promise->cancel();
         }
 
-        $this->emit('end', array($this));
         $this->emit('close', array($this));
     }
 }

tests/UnwrapReadableTest.php

@@ -30,6 +30,9 @@ public function testClosingStreamMakesItNotReadable()
         $promise = new \React\Promise\Promise(function () { });
         $stream = Stream\unwrapReadable($promise);
 
+        $stream->on('close', $this->expectCallableOnce());
+        $stream->on('end', $this->expectCallableNever());
+
         $stream->close();
 
         $this->assertFalse($stream->isReadable());
@@ -54,6 +57,7 @@ public function testEmitsErrorWhenPromiseRejects()
         $this->assertTrue($stream->isReadable());
 
         $stream->on('error', $this->expectCallableOnce());
+        $stream->on('end', $this->expectCallableNever());
 
         $this->loop->run();
 
@@ -69,6 +73,7 @@ public function testEmitsErrorWhenPromiseResolvesWithWrongValue()
         $this->assertTrue($stream->isReadable());
 
         $stream->on('error', $this->expectCallableOnce());
+        $stream->on('end', $this->expectCallableNever());
 
         $this->loop->run();
 
@@ -160,6 +165,7 @@ public function testEmitsCloseOnlyOnceWhenClosingStreamMultipleTimes()
         $promise = new Promise\Promise(function () { });
         $stream = Stream\unwrapReadable($promise);
 
+        $stream->on('end', $this->expectCallableNever());
         $stream->on('close', $this->expectCallableOnce());
 
         $stream->close();