From c353b3fbd86e30b5b51219e8867d479ea11e6811 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E6=9D=8E=E9=93=AD=E6=98=95?= <715557344@qq.com>
Date: Tue, 24 Dec 2024 16:59:48 +0800
Subject: [PATCH] Added `Hyperf\Coroutine\Mutex`. (#7224)

* Added `Hyperf\Coroutine\Mutex`.

* Update CHANGELOG-3.1.md
---
 src/Mutex.php       | 61 +++++++++++++++++++++++++++++++++++++++++++++
 tests/MutexTest.php | 60 ++++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 121 insertions(+)
 create mode 100644 src/Mutex.php
 create mode 100644 tests/MutexTest.php

diff --git a/src/Mutex.php b/src/Mutex.php
new file mode 100644
index 0000000..ed2d3b5
--- /dev/null
+++ b/src/Mutex.php
@@ -0,0 +1,61 @@
+<?php
+
+declare(strict_types=1);
+/**
+ * This file is part of Hyperf.
+ *
+ * @link     https://www.hyperf.io
+ * @document https://hyperf.wiki
+ * @contact  group@hyperf.io
+ * @license  https://github.com/hyperf/hyperf/blob/master/LICENSE
+ */
+
+namespace Hyperf\Coroutine;
+
+use Hyperf\Engine\Channel;
+
+class Mutex
+{
+    /**
+     * @var Channel[]
+     */
+    protected static array $channels = [];
+
+    public static function lock(string $key, float $timeout = -1): bool
+    {
+        if (! isset(static::$channels[$key])) {
+            static::$channels[$key] = new Channel(1);
+        }
+
+        $channel = static::$channels[$key];
+        $channel->push(1, $timeout);
+        if ($channel->isTimeout() || $channel->isClosing()) {
+            return false;
+        }
+
+        return true;
+    }
+
+    public static function unlock(string $key, float $timeout = 5): bool
+    {
+        if (isset(static::$channels[$key])) {
+            $channel = static::$channels[$key];
+            $channel->pop($timeout);
+            if ($channel->isTimeout()) {
+                // unlock more than once
+                return false;
+            }
+        }
+
+        return true;
+    }
+
+    public static function clear(string $key): void
+    {
+        if (isset(static::$channels[$key])) {
+            $channel = static::$channels[$key];
+            static::$channels[$key] = null;
+            $channel->close();
+        }
+    }
+}
diff --git a/tests/MutexTest.php b/tests/MutexTest.php
new file mode 100644
index 0000000..679e458
--- /dev/null
+++ b/tests/MutexTest.php
@@ -0,0 +1,60 @@
+<?php
+
+declare(strict_types=1);
+/**
+ * This file is part of Hyperf.
+ *
+ * @link     https://www.hyperf.io
+ * @document https://hyperf.wiki
+ * @contact  group@hyperf.io
+ * @license  https://github.com/hyperf/hyperf/blob/master/LICENSE
+ */
+
+namespace HyperfTest\Coroutine;
+
+use Hyperf\Coroutine\Mutex;
+use Hyperf\Coroutine\WaitGroup;
+use Hyperf\Engine\Channel;
+use PHPUnit\Framework\Attributes\CoversNothing;
+use PHPUnit\Framework\TestCase;
+
+use function Hyperf\Coroutine\go;
+
+/**
+ * @internal
+ * @coversNothing
+ */
+#[CoversNothing]
+class MutexTest extends TestCase
+{
+    public function testMutexLock()
+    {
+        $chan = new Channel(5);
+        $func = function (string $value) use ($chan) {
+            if (Mutex::lock('test')) {
+                try {
+                    usleep(1000);
+                    $chan->push($value);
+                } finally {
+                    Mutex::unlock('test');
+                }
+            }
+        };
+
+        $wg = new WaitGroup(5);
+        foreach (['h', 'e', 'l', 'l', 'o'] as $value) {
+            go(function () use ($func, $value, $wg) {
+                $func($value);
+                $wg->done();
+            });
+        }
+
+        $res = '';
+        $wg->wait(1);
+        for ($i = 0; $i < 5; ++$i) {
+            $res .= $chan->pop(1);
+        }
+
+        $this->assertSame('hello', $res);
+    }
+}