Add ThroughStream and ReadableStream primitives

These are basic building blocks for building custom streams upon. The
ThroughStream is essentially a filter which allows you to change the value
or perform a passive action (like logging) by overriding the filter()
method.

Example:

    class FooStream extends ThroughStream
    {
        public function filter($data)
        {
            return foo.$data;
        }
    }

This useful example prepends the string "foo" to any data it receives.

Author: igorw

ReadableStream.php

@@ -0,0 +1,42 @@
+<?php
+
+namespace React\Stream;
+
+use Evenement\EventEmitter;
+
+class ReadableStream extends EventEmitter implements ReadableStreamInterface
+{
+    public $closed = false;
+
+    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));
+        $this->removeAllListeners();
+    }
+}

ThroughStream.php

@@ -0,0 +1,85 @@
+<?php
+
+namespace React\Stream;
+
+use Evenement\EventEmitter;
+
+class ThroughStream extends EventEmitter implements ReadableStreamInterface, WritableStreamInterface
+{
+    private $closed = false;
+    private $pipeSource;
+
+    public function __construct()
+    {
+        $this->on('pipe', array($this, 'handlePipeEvent'));
+    }
+
+    public function handlePipeEvent($source)
+    {
+        $this->pipeSource = $source;
+    }
+
+    public function filter($data)
+    {
+        return $data;
+    }
+
+    public function write($data)
+    {
+        $this->emit('data', array($this->filter($data)));
+    }
+
+    public function end($data = null)
+    {
+        if (null !== $data) {
+            $this->write($data);
+        }
+
+        $this->close();
+    }
+
+    public function isReadable()
+    {
+        return !$this->closed;
+    }
+
+    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->closed = true;
+        $this->pipeSource = null;
+
+        $this->emit('end');
+        $this->emit('close');
+        $this->removeAllListeners();
+    }
+
+    public function pipe(WritableStreamInterface $dest, array $options = array())
+    {
+        Util::pipe($this, $dest, $options);
+
+        return $dest;
+    }
+}