Refactor to use internal class

Author: clue

composer.json

@@ -11,6 +11,7 @@
         }
     ],
     "autoload": {
+        "psr-4": { "Clue\\React\\Promise\\Stream\\" : "src/" },
         "files": [ "src/functions.php" ]
     },
     "require": {

src/UnwrapReadableStream.php

@@ -0,0 +1,112 @@
+<?php
+
+namespace Clue\React\Promise\Stream;
+
+use Evenement\EventEmitter;
+use React\Stream\ReadableStreamInterface;
+use React\Promise\PromiseInterface;
+use React\Stream\WritableStreamInterface;
+use React\Stream\Util;
+use React\Promise\CancellablePromiseInterface;
+
+/**
+ * @internal
+ * @see unwrapReadable() instead
+ */
+class UnwrapReadableStream extends EventEmitter implements ReadableStreamInterface
+{
+    private $promise;
+    private $closed = false;
+
+    /**
+     * unwrap a `Promise` which resolves with a `ReadableStreamInterface`.
+     *
+     * @param PromiseInterface $promise Promise<ReadableStreamInterface, Exception>
+     * @return ReadableStreamInterface
+     */
+    public function __construct(PromiseInterface $promise)
+    {
+        $out = $this;
+
+        // TODO: support backpressure
+
+        // try to cancel promise once the stream closes
+        if ($promise instanceof CancellablePromiseInterface) {
+            $out->on('close', function() use ($promise) {
+                $promise->cancel();
+            });
+        }
+
+        $this->promise = $promise->then(
+            function ($stream) {
+                if (!($stream instanceof ReadableStreamInterface)) {
+                    throw new \InvalidArgumentException('Not a readable stream');
+                }
+                return $stream;
+            }
+        )->then(
+            function (ReadableStreamInterface $stream) use ($out) {
+                if (!$stream->isReadable()) {
+                    $out->close();
+                    return $stream;
+                }
+
+                // stream any writes into output stream
+                $stream->on('data', function ($data) use ($out) {
+                    $out->emit('data', array($data, $out));
+                });
+
+                // 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
+                $stream->on('close', function () use ($out) {
+                    $out->close();
+                });
+                $stream->on('end', function () use ($out) {
+                    $out->close();
+                });
+                return $stream;
+            },
+            function ($e) use ($out) {
+                $out->emit('error', array($e, $out));
+                $out->close();
+            }
+        );
+    }
+
+    public function isReadable()
+    {
+        return !$this->closed;
+    }
+
+    public function pause()
+    {
+    }
+
+    public function resume()
+    {
+    }
+
+    public function pipe(WritableStreamInterface $dest, array $options = array())
+    {
+        Util::pipe($this, $dest, $options);
+
+        return $dest;
+    }
+
+    public function close()
+    {
+        if ($this->closed) {
+            return;
+        }
+
+        $this->closed = true;
+
+        $this->emit('end', array($this));
+        $this->emit('close', array($this));
+    }
+}

src/functions.php

@@ -56,55 +56,5 @@ function buffer(ReadableStreamInterface $stream)
  */
 function unwrapReadable(PromiseInterface $promise)
 {
-    $out = new ReadableStream();
-
-    // TODO: support backpressure
-
-    // try to cancel promise once the stream closes
-    if ($promise instanceof CancellablePromiseInterface) {
-        $out->on('close', function() use ($promise) {
-            $promise->cancel();
-        });
-    }
-
-    $promise->then(
-        function ($stream) {
-            if (!($stream instanceof ReadableStreamInterface)) {
-                throw new \InvalidArgumentException('Not a readable stream');
-            }
-            return $stream;
-        }
-    )->then(
-        function (ReadableStreamInterface $stream) use ($out) {
-            if (!$stream->isReadable()) {
-                $out->close();
-                return;
-            }
-
-            // stream any writes into output stream
-            $stream->on('data', function ($data) use ($out) {
-                $out->emit('data', array($data, $out));
-            });
-
-            // 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
-            $stream->on('close', function () use ($out) {
-                $out->close();
-            });
-            $stream->on('end', function () use ($out) {
-                $out->close();
-            });
-        },
-        function ($e) use ($out) {
-            $out->emit('error', array($e, $out));
-            $out->close();
-        }
-    );
-
-    return $out;
+    return new UnwrapReadableStream($promise);
 }