Add setTimeout & Cancel API

src/utils/src/Context.php

@@ -12,10 +12,13 @@ declare(strict_types=1);
 
 namespace Hyperf\Utils;
 
+use Carbon\Carbon;
 use Swoole\Coroutine as SwCoroutine;
 
 class Context
 {
+    public const DONE = 'hyperf.context.done';
+
     protected static $nonCoContext = [];
 
     public static function set(string $id, $value)
@@ -108,4 +111,50 @@ class Context
 
         return static::$nonCoContext;
     }
+
+    public static function done(): bool
+    {
+        $holder = self::getOrSet(static::DONE, new \stdClass());
+        return $holder->done ?? false;
+    }
+
+    /**
+     * @param float|int $millisecond
+     */
+    public static function setTimeout($millisecond)
+    {
+        $holder = self::getOrSet(static::DONE, new \stdClass());
+        Coroutine::create(function () use ($holder, $millisecond) {
+            usleep((int) $millisecond * 1000);
+            $holder->done = true;
+        });
+    }
+
+    public static function setDeadline(\DateTime $deadline)
+    {
+        if (! ($deadline instanceof Carbon)) {
+            $deadline = Carbon::instance($deadline);
+        }
+
+        $timeout = $deadline->getPreciseTimestamp(3) - microtime(true) * 1000;
+        $timeout = $timeout > 0 ? $timeout : 0;
+        static::setTimeout($timeout);
+    }
+
+    public static function cancel(): void
+    {
+        $holder = self::getOrSet(static::DONE, new \stdClass());
+        $holder->done = true;
+    }
+
+    public static function go(callable $callable): int
+    {
+        if (! self::has(static::DONE)) {
+            self::set(static::DONE, new \stdClass());
+        }
+        return Coroutine::create(function () use ($callable) {
+            static::copy(coroutine::parentId());
+            $callable();
+        });
+    }
 }

src/utils/tests/ContextTest.php

@@ -12,8 +12,12 @@ declare(strict_types=1);
 
 namespace HyperfTest\Utils;
 
+use Carbon\Carbon;
 use Hyperf\Utils\Context;
 use PHPUnit\Framework\TestCase;
+use Swoole\Coroutine\Channel;
+use Swoole\Coroutine\System;
+use Swoole\Runtime;
 
 /**
  * @internal
@@ -45,4 +49,101 @@ class ContextTest extends TestCase
         Context::set('test.store.id', null);
         $this->assertSame(1, Context::getOrSet('test.store.id', 1));
     }
+
+    public function testCancel()
+    {
+        Context::set(Context::DONE, null);
+        $chan = new Channel(1);
+        Context::go(function () use ($chan) {
+            if (Context::done()) {
+                $chan->push(1);
+                return;
+            }
+            $chan->push(2);
+        });
+        $this->assertEquals(2, $chan->pop());
+        Context::cancel();
+        Context::go(function () use ($chan) {
+            if (Context::done()) {
+                $chan->push(1);
+                return;
+            }
+            $chan->push(2);
+        });
+        $this->assertEquals(1, $chan->pop());
+    }
+
+    public function testNestedCancel()
+    {
+        Context::set(Context::DONE, null);
+        $chan = new Channel(1);
+        Context::go(function () use ($chan) {
+            if (Context::done()) {
+                $chan->push(1);
+                return;
+            }
+            usleep(20000);
+            System::sleep(1);
+            Context::go(function () use ($chan) {
+                if (Context::done()) {
+                    $chan->push(2);
+                    return;
+                }
+                $chan->push(3);
+            });
+        });
+        usleep(10000);
+        Context::cancel();
+        $this->assertEquals(2, $chan->pop());
+    }
+
+    public function testTimeout()
+    {
+        Context::set(Context::DONE, null);
+        Runtime::enableCoroutine();
+        Context::setTimeout(5);
+        $chan = new Channel(1);
+        Context::go(function () use ($chan) {
+            if (Context::done()) {
+                $chan->push(1);
+                return;
+            }
+            $chan->push(2);
+        });
+        $this->assertEquals(2, $chan->pop());
+        Context::go(function () use ($chan) {
+            usleep(10000);
+            if (Context::done()) {
+                $chan->push(1);
+                return;
+            }
+            $chan->push(2);
+        });
+        $this->assertEquals(1, $chan->pop());
+    }
+
+    public function testDeadline()
+    {
+        Context::set(Context::DONE, null);
+        $deadline = Carbon::now()->addMillisecond(5);
+        Context::setDeadline($deadline);
+        $chan = new Channel(1);
+        Context::go(function () use ($chan) {
+            if (Context::done()) {
+                $chan->push(1);
+                return;
+            }
+            $chan->push(2);
+        });
+        $this->assertEquals(2, $chan->pop());
+        usleep(10000);
+        Context::go(function () use ($chan) {
+            if (Context::done()) {
+                $chan->push(1);
+                return;
+            }
+            $chan->push(2);
+        });
+        $this->assertEquals(1, $chan->pop());
+    }
 }