Consistent event arguments and documentation

README.md

@@ -18,9 +18,11 @@ This component depends on `événement`, which is an implementation of the
 
 ### EventEmitter Events
 
-* `data`: Emitted whenever data was read from the source.
+* `data`: Emitted whenever data was read from the source
+  with a single mixed argument for incoming data.
 * `end`: Emitted when the source has reached the `eof`.
-* `error`: Emitted when an error occurs.
+* `error`: Emitted when an error occurs
+  with a single `Exception` argument for error instance.
 * `close`: Emitted when the connection is closed.
 
 ### Methods
@@ -42,9 +44,11 @@ This component depends on `événement`, which is an implementation of the
 
 * `drain`: Emitted if the write buffer became full previously and is now ready
   to accept more data.
-* `error`: Emitted whenever an error occurs.
+* `error`: Emitted whenever an error occurs
+  with a single `Exception` argument for error instance.
 * `close`: Emitted whenever the connection is closed.
-* `pipe`: Emitted whenever a readable stream is `pipe()`d into this stream.
+* `pipe`: Emitted whenever a readable stream is `pipe()`d into this stream
+  with a single `ReadableStreamInterface` argument for source stream.
 
 ### Methods
 

src/Buffer.php

@@ -117,7 +117,7 @@ public function handleWrite()
                 );
             }
 
-            $this->emit('error', array(new \RuntimeException('Unable to write to stream: ' . $error->getMessage(), 0, $error), $this));
+            $this->emit('error', array(new \RuntimeException('Unable to write to stream: ' . $error->getMessage(), 0, $error)));
             $this->close();
 
             return;
@@ -128,7 +128,7 @@ public function handleWrite()
 
         // buffer has been above limit and is now below limit
         if ($exceeded && !isset($this->data[$this->softLimit - 1])) {
-            $this->emit('drain', array($this));
+            $this->emit('drain');
         }
 
         // buffer is now completely empty => stop trying to write

src/ReadableStream.php

@@ -35,8 +35,8 @@ public function close()
         }
 
         $this->closed = true;
-        $this->emit('end', array($this));
-        $this->emit('close', array($this));
+        $this->emit('end');
+        $this->emit('close');
         $this->removeAllListeners();
     }
 }

src/ReadableStreamInterface.php

@@ -5,9 +5,9 @@
 use Evenement\EventEmitterInterface;
 
 /**
- * @event data
+ * @event data with a single mixed argument for incoming data
  * @event end
- * @event error
+ * @event error with a single Exception argument for error instance
  * @event close
  */
 interface ReadableStreamInterface extends EventEmitterInterface

src/Stream.php

@@ -62,13 +62,13 @@ public function __construct($stream, LoopInterface $loop, WritableStreamInterfac
         $that = $this;
 
         $this->buffer->on('error', function ($error) use ($that) {
-            $that->emit('error', array($error, $that));
+            $that->emit('error', array($error));
         });
 
         $this->buffer->on('close', array($this, 'close'));
 
         $this->buffer->on('drain', function () use ($that) {
-            $that->emit('drain', array($that));
+            $that->emit('drain');
         });
 
         $this->resume();
@@ -116,8 +116,8 @@ public function close()
         $this->readable = false;
         $this->writable = false;
 
-        $this->emit('end', array($this));
-        $this->emit('close', array($this));
+        $this->emit('end');
+        $this->emit('close');
         $this->loop->removeStream($this->stream);
         $this->buffer->close();
         $this->removeAllListeners();
@@ -164,13 +164,13 @@ public function handleData($stream)
         restore_error_handler();
 
         if ($error !== null) {
-            $this->emit('error', array(new \RuntimeException('Unable to read from stream: ' . $error->getMessage(), 0, $error), $this));
+            $this->emit('error', array(new \RuntimeException('Unable to read from stream: ' . $error->getMessage(), 0, $error)));
             $this->close();
             return;
         }
 
         if ($data !== '') {
-            $this->emit('data', array($data, $this));
+            $this->emit('data', array($data));
         }
 
         if (!is_resource($stream) || feof($stream)) {

src/ThroughStream.php

@@ -19,13 +19,13 @@ public function filter($data)
 
     public function write($data)
     {
-        $this->readable->emit('data', array($this->filter($data), $this));
+        $this->readable->emit('data', array($this->filter($data)));
     }
 
     public function end($data = null)
     {
         if (null !== $data) {
-            $this->readable->emit('data', array($this->filter($data), $this));
+            $this->readable->emit('data', array($this->filter($data)));
         }
 
         $this->writable->end($data);

src/WritableStream.php

@@ -33,8 +33,8 @@ public function close()
         }
 
         $this->closed = true;
-        $this->emit('end', array($this));
-        $this->emit('close', array($this));
+        $this->emit('end');
+        $this->emit('close');
         $this->removeAllListeners();
     }
 }

src/WritableStreamInterface.php

@@ -6,9 +6,9 @@
 
 /**
  * @event drain
- * @event error
+ * @event error with a single Exeption argument for error instance
  * @event close
- * @event pipe
+ * @event pipe with a single ReadableStreamInterface argument for source stream
  */
 interface WritableStreamInterface extends EventEmitterInterface
 {

tests/BufferTest.php

@@ -186,7 +186,7 @@ public function testWriteInDrain()
         $buffer->softLimit = 2;
         $buffer->on('error', $this->expectCallableNever());
 
-        $buffer->once('drain', function ($buffer) {
+        $buffer->once('drain', function () use ($buffer) {
             $buffer->listening = false;
             $buffer->write("bar\n");
         });

tests/StreamIntegrationTest.php

@@ -40,7 +40,7 @@ public function testBufferReadsLargeChunks($condition, $loopFactory)
         $testString = str_repeat("*", $streamA->bufferSize + 1);
 
         $buffer = "";
-        $streamB->on('data', function ($data, $streamB) use (&$buffer, &$testString) {
+        $streamB->on('data', function ($data) use (&$buffer) {
             $buffer .= $data;
         });
 

tests/StreamTest.php

@@ -166,7 +166,7 @@ public function testEnd()
     }
 
     public function testEndedStreamsShouldNotWrite()
-    {   
+    {
         $file = tempnam(sys_get_temp_dir(), 'reactphptest_');
         $stream = fopen($file, 'r+');
         $loop = $this->createWriteableLoopMock();
@@ -208,8 +208,8 @@ public function testClosingStreamInDataEventShouldNotTriggerError()
         $loop = $this->createLoopMock();
 
         $conn = new Stream($stream, $loop);
-        $conn->on('data', function ($data, $stream) {
-            $stream->close();
+        $conn->on('data', function ($data) use ($conn) {
+            $conn->close();
         });
 
         fwrite($stream, "foobar\n");