Data callback is allowed to throw Exception

Author: clue

README.md

@@ -972,6 +972,23 @@ $through->on('data', $this->expectCallableOnceWith("[2, true]\n"));
 $through->write(array(2, true));
 ```
 
+The callback function is allowed to throw an `Exception`. In this case,
+the stream will emit an `error` event and then [`close()`](#close-1) the stream.
+
+```php
+$through = new ThroughStream(function ($data) {
+    if (!is_string($data)) {
+        throw new \UnexpectedValueException('Only strings allowed');
+    }
+    return $data;
+});
+$through->on('error', $this->expectCallableOnce()));
+$through->on('close', $this->expectCallableOnce()));
+$through->on('data', $this->expectCallableNever()));
+
+$through->write(2);
+```
+
 ## Usage
 ```php
     $loop = React\EventLoop\Factory::create();

src/ThroughStream.php

@@ -51,6 +51,23 @@
  * $through->write(array(2, true));
  * ```
  *
+ * The callback function is allowed to throw an `Exception`. In this case,
+ * the stream will emit an `error` event and then [`close()`](#close-1) the stream.
+ *
+ * ```php
+ * $through = new ThroughStream(function ($data) {
+ *     if (!is_string($data)) {
+ *         throw new \UnexpectedValueException('Only strings allowed');
+ *     }
+ *     return $data;
+ * });
+ * $through->on('error', $this->expectCallableOnce()));
+ * $through->on('close', $this->expectCallableOnce()));
+ * $through->on('data', $this->expectCallableNever()));
+ *
+ * $through->write(2);
+ * ```
+ *
  * @see WritableStreamInterface::write()
  * @see WritableStreamInterface::end()
  * @see DuplexStreamInterface::close()
@@ -109,7 +126,18 @@ public function write($data)
             return false;
         }
 
-        $this->emit('data', array($this->filter($data)));
+        if ($this->callback !== null) {
+            try {
+                $data = call_user_func($this->callback, $data);
+            } catch (\Exception $e) {
+                $this->emit('error', array($e));
+                $this->close();
+
+                return false;
+            }
+        }
+
+        $this->emit('data', array($data));
 
         if ($this->paused) {
             $this->drain = true;
@@ -127,6 +155,11 @@ public function end($data = null)
 
         if (null !== $data) {
             $this->write($data);
+
+            // return if write() already caused the stream to close
+            if (!$this->writable) {
+                return;
+            }
         }
 
         $this->readable = false;
@@ -153,13 +186,4 @@ public function close()
 
         $this->emit('close');
     }
-
-    private function filter($data)
-    {
-        if ($this->callback !== null) {
-            $data = call_user_func($this->callback, $data);
-        }
-
-        return $data;
-    }
 }

tests/ThroughStreamTest.php

@@ -52,6 +52,40 @@ public function itShouldEmitAnyDataWrittenToItPassedThruCallback()
         $through->write('foo');
     }
 
+    /** @test */
+    public function itShouldEmitErrorAndCloseIfCallbackThrowsException()
+    {
+        $through = new ThroughStream(function () {
+            throw new \RuntimeException();
+        });
+        $through->on('error', $this->expectCallableOnce());
+        $through->on('close', $this->expectCallableOnce());
+        $through->on('data', $this->expectCallableNever());
+        $through->on('end', $this->expectCallableNever());
+
+        $through->write('foo');
+
+        $this->assertFalse($through->isReadable());
+        $this->assertFalse($through->isWritable());
+    }
+
+    /** @test */
+    public function itShouldEmitErrorAndCloseIfCallbackThrowsExceptionOnEnd()
+    {
+        $through = new ThroughStream(function () {
+            throw new \RuntimeException();
+        });
+        $through->on('error', $this->expectCallableOnce());
+        $through->on('close', $this->expectCallableOnce());
+        $through->on('data', $this->expectCallableNever());
+        $through->on('end', $this->expectCallableNever());
+
+        $through->end('foo');
+
+        $this->assertFalse($through->isReadable());
+        $this->assertFalse($through->isWritable());
+    }
+
     /** @test */
     public function itShouldReturnFalseForAnyDataWrittenToItWhenPaused()
     {