Merge pull request #36 from lucasnetau/gc_cycles

Improve `first()` promise resolution to clean up any garbage references

Author: clue

src/functions.php

@@ -141,6 +141,7 @@ function first(EventEmitterInterface $stream, $event = 'data')
     return new Promise\Promise(function ($resolve, $reject) use ($stream, $event, &$listener) {
         $listener = function ($data = null) use ($stream, $event, &$listener, $resolve) {
             $stream->removeListener($event, $listener);
+            $listener = null;
             $resolve($data);
         };
         $stream->on($event, $listener);
@@ -156,8 +157,11 @@ function first(EventEmitterInterface $stream, $event = 'data')
             });
         }
 
-        $stream->on('close', function () use ($stream, $event, $listener, $reject) {
-            $stream->removeListener($event, $listener);
+        $stream->on('close', function () use ($stream, $event, &$listener, $reject) {
+            if ($listener !== null) {
+                $stream->removeListener($event, $listener);
+                $listener = null;
+            }
             $reject(new \RuntimeException('Stream closed'));
         });
     }, function ($_, $reject) use ($stream, $event, &$listener) {

tests/FirstTest.php

@@ -112,4 +112,43 @@ public function testCancelPendingStreamWillReject()
 
         $this->expectPromiseReject($promise);
     }
+
+    public function testNoGarbageCollectionCyclesAfterClosingStream()
+    {
+        \gc_collect_cycles();
+        $stream = new ThroughStream();
+        $promise = Stream\first($stream);
+
+        $stream->close();
+
+        $this->assertSame(0, \gc_collect_cycles());
+    }
+
+    public function testShouldResolveWithoutCreatingGarbageCyclesAfterDataThenClose()
+    {
+        \gc_collect_cycles();
+
+        $stream = new ThroughStream();
+
+        $promise = Stream\first($stream);
+
+        $stream->emit('data', array('hello', $stream));
+        $stream->close();
+
+        $this->expectPromiseResolve($promise);
+        $this->assertSame(0, \gc_collect_cycles());
+    }
+
+    public function testCancelPendingStreamWillRejectWithoutCreatingGarbageCycles()
+    {
+        \gc_collect_cycles();
+        $stream = new ThroughStream();
+
+        $promise = Stream\first($stream);
+
+        $promise->cancel();
+
+        $this->expectPromiseReject($promise);
+        $this->assertSame(0, \gc_collect_cycles());
+    }
 }