Add CompositeStream, make ThroughStream extend it

Author: igorw

CompositeStream.php

@@ -0,0 +1,95 @@
+<?php
+
+namespace React\Stream;
+
+use Evenement\EventEmitter;
+
+class CompositeStream extends EventEmitter implements ReadableStreamInterface, WritableStreamInterface
+{
+    protected $readable;
+    protected $writable;
+    protected $pipeSource;
+
+    public function __construct(ReadableStreamInterface $readable, WritableStreamInterface $writable)
+    {
+        $this->readable = $readable;
+        $this->writable = $writable;
+
+        $this->forwardEvents($this->readable, ['data', 'end', 'error', 'close']);
+        $this->forwardEvents($this->writable, ['drain', 'error', 'close', 'pipe']);
+
+        $this->readable->on('close', array($this, 'close'));
+        $this->writable->on('close', array($this, 'close'));
+
+        $this->on('pipe', array($this, 'handlePipeEvent'));
+    }
+
+    public function handlePipeEvent($source)
+    {
+        $this->pipeSource = $source;
+    }
+
+    public function isReadable()
+    {
+        return $this->readable->isReadable();
+    }
+
+    public function pause()
+    {
+        if ($this->pipeSource) {
+            $this->pipeSource->pause();
+        }
+
+        $this->readable->pause();
+    }
+
+    public function resume()
+    {
+        if ($this->pipeSource) {
+            $this->pipeSource->resume();
+        }
+
+        $this->readable->resume();
+    }
+
+    public function pipe(WritableStreamInterface $dest, array $options = array())
+    {
+        Util::pipe($this, $dest, $options);
+
+        return $dest;
+    }
+
+    public function isWritable()
+    {
+        return $this->writable->isWritable();
+    }
+
+    public function write($data)
+    {
+        return $this->writable->write($data);
+    }
+
+    public function end($data = null)
+    {
+        $this->writable->end($data);
+    }
+
+    public function close()
+    {
+        $this->pipeSource = true;
+
+        $this->readable->close();
+        $this->writable->close();
+    }
+
+    protected function forwardEvents($stream, array $events)
+    {
+        $that = $this;
+
+        foreach ($events as $event) {
+            $stream->on($event, function () use ($event, $that) {
+                $that->emit($event, func_get_args());
+            });
+        }
+    }
+}

ThroughStream.php

@@ -2,18 +2,14 @@
 
 namespace React\Stream;
 
-class ThroughStream extends ReadableStream implements WritableStreamInterface
+class ThroughStream extends CompositeStream
 {
-    private $pipeSource;
-
     public function __construct()
     {
-        $this->on('pipe', array($this, 'handlePipeEvent'));
-    }
+        $readable = new ReadableStream();
+        $writable = new WritableStream();
 
-    public function handlePipeEvent($source)
-    {
-        $this->pipeSource = $source;
+        parent::__construct($readable, $writable);
     }
 
     public function filter($data)
@@ -23,44 +19,15 @@ public function filter($data)
 
     public function write($data)
     {
-        $this->emit('data', array($this->filter($data)));
+        $this->readable->emit('data', array($this->filter($data)));
     }
 
     public function end($data = null)
     {
         if (null !== $data) {
-            $this->write($data);
-        }
-
-        $this->close();
-    }
-
-    public function isWritable()
-    {
-        return !$this->closed;
-    }
-
-    public function pause()
-    {
-        if ($this->pipeSource) {
-            $this->pipeSource->pause();
-        }
-    }
-
-    public function resume()
-    {
-        if ($this->pipeSource) {
-            $this->pipeSource->resume();
-        }
-    }
-
-    public function close()
-    {
-        if ($this->closed) {
-            return;
+            $this->readable->emit('data', array($this->filter($data)));
         }
 
-        parent::close();
-        $this->pipeSource = null;
+        $this->writable->end($data);
     }
 }

WritableStream.php

@@ -7,17 +7,6 @@
 class WritableStream extends EventEmitter implements WritableStreamInterface
 {
     public $closed = false;
-    private $pipeSource;
-
-    public function __construct()
-    {
-        $this->on('pipe', array($this, 'handlePipeEvent'));
-    }
-
-    public function handlePipeEvent($source)
-    {
-        $this->pipeSource = $source;
-    }
 
     public function write($data)
     {