Global queue

src/FulfilledPromise.php

@@ -21,13 +21,17 @@ public function then(callable $onFulfilled = null, callable $onRejected = null,
             return $this;
         }
 
-        try {
-            return resolve($onFulfilled($this->value));
-        } catch (\Throwable $exception) {
-            return new RejectedPromise($exception);
-        } catch (\Exception $exception) {
-            return new RejectedPromise($exception);
-        }
+        return new Promise(function (callable $resolve, callable $reject) use ($onFulfilled) {
+            queue()->enqueue(function () use ($resolve, $reject, $onFulfilled) {
+                try {
+                    $resolve($onFulfilled($this->value));
+                } catch (\Throwable $exception) {
+                    $reject($exception);
+                } catch (\Exception $exception) {
+                    $reject($exception);
+                }
+            });
+        });
     }
 
     public function done(callable $onFulfilled = null, callable $onRejected = null, callable $onProgress = null)
@@ -36,11 +40,13 @@ public function done(callable $onFulfilled = null, callable $onRejected = null,
             return;
         }
 
-        $result = $onFulfilled($this->value);
+        queue()->enqueue(function () use ($onFulfilled) {
+            $result = $onFulfilled($this->value);
 
-        if ($result instanceof ExtendedPromiseInterface) {
-            $result->done();
-        }
+            if ($result instanceof ExtendedPromiseInterface) {
+                $result->done();
+            }
+        });
     }
 
     public function otherwise(callable $onRejected)

src/Promise.php

@@ -148,20 +148,24 @@ private function notify($update = null)
             return;
         }
 
-        foreach ($this->progressHandlers as $handler) {
-            $handler($update);
-        }
+        $handlers = $this->progressHandlers;
+
+        queue()->enqueue(function () use ($handlers, $update) {
+            foreach ($handlers as $handler) {
+                $handler($update);
+            }
+        });
     }
 
-    private function settle(ExtendedPromiseInterface $promise)
+    private function settle(ExtendedPromiseInterface $result)
     {
         $handlers = $this->handlers;
 
         $this->progressHandlers = $this->handlers = [];
-        $this->result = $promise;
+        $this->result = $result;
 
         foreach ($handlers as $handler) {
-            $handler($promise);
+            $handler($result);
         }
     }
 

src/Queue/QueueInterface.php

@@ -0,0 +1,8 @@
+<?php
+
+namespace React\Promise\Queue;
+
+interface QueueInterface
+{
+    public function enqueue(callable $task);
+}

src/Queue/SynchronousQueue.php

@@ -0,0 +1,38 @@
+<?php
+
+namespace React\Promise\Queue;
+
+class SynchronousQueue implements QueueInterface
+{
+    private $queue = [];
+
+    public function enqueue(callable $task)
+    {
+        if (1 === array_push($this->queue, $task)) {
+            $this->drain();
+        }
+    }
+
+    private function drain()
+    {
+        for ($i = key($this->queue); isset($this->queue[$i]); $i++) {
+            $task = $this->queue[$i];
+
+            $exception = null;
+
+            try {
+                $task();
+            } catch (\Throwable $exception) {
+            } catch (\Exception $exception) {
+            }
+
+            unset($this->queue[$i]);
+
+            if ($exception) {
+                throw $exception;
+            }
+        }
+
+        $this->queue = [];
+    }
+}

src/RejectedPromise.php

@@ -21,30 +21,36 @@ public function then(callable $onFulfilled = null, callable $onRejected = null,
             return $this;
         }
 
-        try {
-            return resolve($onRejected($this->reason));
-        } catch (\Throwable $exception) {
-            return new RejectedPromise($exception);
-        } catch (\Exception $exception) {
-            return new RejectedPromise($exception);
-        }
+        return new Promise(function (callable $resolve, callable $reject) use ($onRejected) {
+            queue()->enqueue(function () use ($resolve, $reject, $onRejected) {
+                try {
+                    $resolve($onRejected($this->reason));
+                } catch (\Throwable $exception) {
+                    $reject($exception);
+                } catch (\Exception $exception) {
+                    $reject($exception);
+                }
+            });
+        });
     }
 
     public function done(callable $onFulfilled = null, callable $onRejected = null, callable $onProgress = null)
     {
-        if (null === $onRejected) {
-            throw UnhandledRejectionException::resolve($this->reason);
-        }
+        queue()->enqueue(function () use ($onRejected) {
+            if (null === $onRejected) {
+                throw UnhandledRejectionException::resolve($this->reason);
+            }
 
-        $result = $onRejected($this->reason);
+            $result = $onRejected($this->reason);
 
-        if ($result instanceof self) {
-            throw UnhandledRejectionException::resolve($result->reason);
-        }
+            if ($result instanceof self) {
+                throw UnhandledRejectionException::resolve($result->reason);
+            }
 
-        if ($result instanceof ExtendedPromiseInterface) {
-            $result->done();
-        }
+            if ($result instanceof ExtendedPromiseInterface) {
+                $result->done();
+            }
+        });
     }
 
     public function otherwise(callable $onRejected)

src/functions.php

@@ -223,6 +223,21 @@ function reduce($promisesOrValues, callable $reduceFunc, $initialValue = null)
     }, $cancellationQueue);
 }
 
+function queue(Queue\QueueInterface $queue = null)
+{
+    static $globalQueue;
+
+    if ($queue) {
+        return ($globalQueue = $queue);
+    }
+
+    if (!$globalQueue) {
+        $globalQueue = new Queue\SynchronousQueue();
+    }
+
+    return $globalQueue;
+}
+
 // Internal functions
 function _checkTypehint(callable $callback, $object)
 {

tests/FunctionResolveTest.php

@@ -126,13 +126,12 @@ public function shouldSupportVeryDeepNestedPromises()
     {
         $deferreds = [];
 
-        // @TODO Increase count once global-queue is merged
-        for ($i = 0; $i < 10; $i++) {
+        for ($i = 0; $i < 250; $i++) {
             $deferreds[] = $d = new Deferred();
             $p = $d->promise();
 
             $last = $p;
-            for ($j = 0; $j < 10; $j++) {
+            for ($j = 0; $j < 250; $j++) {
                 $last = $last->then(function($result) {
                     return $result;
                 });