Merge pull request reactphp#246 from jmalloc/stream-drain-event

[Stream] Pass the stream as final parameter on 'drain'.

Author: igorw

src/React/Stream/Buffer.php

@@ -78,13 +78,13 @@ public function close()
         $this->listening = false;
         $this->data = '';
 
-        $this->emit('close');
+        $this->emit('close', [$this]);
     }
 
     public function handleWrite()
     {
         if (!is_resource($this->stream) || ('generic_socket' === $this->meta['stream_type'] && feof($this->stream))) {
-            $this->emit('error', array(new \RuntimeException('Tried to write to closed or invalid stream.')));
+            $this->emit('error', array(new \RuntimeException('Tried to write to closed or invalid stream.'), $this));
 
             return;
         }
@@ -96,20 +96,23 @@ public function handleWrite()
         restore_error_handler();
 
         if (false === $sent) {
-            $this->emit('error', array(new \ErrorException(
-                $this->lastError['message'],
-                0,
-                $this->lastError['number'],
-                $this->lastError['file'],
-                $this->lastError['line']
-            )));
+            $this->emit('error', array(
+                new \ErrorException(
+                    $this->lastError['message'],
+                    0,
+                    $this->lastError['number'],
+                    $this->lastError['file'],
+                    $this->lastError['line']
+                ),
+                $this
+            ));
 
             return;
         }
 
         $len = strlen($this->data);
         if ($len >= $this->softLimit && $len - $sent < $this->softLimit) {
-            $this->emit('drain');
+            $this->emit('drain', [$this]);
         }
 
         $this->data = (string) substr($this->data, $sent);
@@ -118,7 +121,7 @@ public function handleWrite()
             $this->loop->removeWriteStream($this->stream);
             $this->listening = false;
 
-            $this->emit('full-drain');
+            $this->emit('full-drain', [$this]);
         }
     }
 

src/React/Stream/Stream.php

@@ -27,7 +27,7 @@ public function __construct($stream, LoopInterface $loop)
         });
 
         $this->buffer->on('drain', function () {
-            $this->emit('drain');
+            $this->emit('drain', array($this));
         });
 
         $this->resume();

src/React/Stream/ThroughStream.php

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