[WIP] Solve memory leak in Promise::fromObservable

src/React/Promise.php

@@ -46,26 +46,32 @@ public static function rejected($exception): ReactPromise
      */
     public static function fromObservable(ObservableInterface $observable, Deferred $deferred = null): ReactPromise
     {
+        $subscription = null;
+        $promise = new ReactPromise(static function ($resolve, $reject) use (&$subscription, $observable) {
+            $value = null;
+            $subscription = $observable->subscribe(
+                static function ($v) use (&$value) {
+                    $value = $v;
+                },
+                static function ($error) use ($reject) {
+                    $reject($error);
+                },
+                static function () use ($resolve, &$value) {
+                    $resolve($value);
+                }
+            );
 
-        $d = $deferred ?: new Deferred(function () use (&$subscription) {
+        }, static function () use (&$subscription) {
             $subscription->dispose();
         });
 
-        $value = null;
+        if ($deferred === null) {
+            return $promise;
+        }
 
-        $subscription = $observable->subscribe(
-            function ($v) use (&$value) {
-                $value = $v;
-            },
-            function ($error) use ($d) {
-                $d->reject($error);
-            },
-            function () use ($d, &$value) {
-                $d->resolve($value);
-            }
-        );
+        $promise->done([$deferred, 'resolve'], [$deferred, 'reject']);
 
-        return $d->promise();
+        return $deferred->promise();
     }
 
     /**

test/Rx/Functional/React/PromiseFromObservableTest.php

@@ -4,6 +4,7 @@
 namespace Rx\Functional\React;
 
 use Exception;
+use React\Promise\Deferred;
 use Rx\Functional\FunctionalTestCase;
 use Rx\Observable;
 use Rx\React\Promise;
@@ -49,4 +50,31 @@ function ($error) {
               $this->assertEquals($error, new Exception("some error"));
           });
     }
+
+    /**
+     * @test
+     */
+    public function promise_no_memory_leak()
+    {
+        gc_collect_cycles();
+
+        $source = Observable::of(42);
+
+        Promise::fromObservable($source)->done();
+
+        $this->assertSame(0, gc_collect_cycles());
+    }
+    /**
+     * @test
+     */
+    public function promise_no_memory_leak_deferred()
+    {
+        gc_collect_cycles();
+
+        $source = Observable::of(42);
+
+        Promise::fromObservable($source, new Deferred())->done();
+
+        $this->assertSame(0, gc_collect_cycles());
+    }
 }