Wait sometimes until channel manager is empty, before close by reac…

…hed `max_requests`. (#3)

.php-cs-fixer.php

@@ -1,5 +1,17 @@
 <?php
 
+declare(strict_types=1);
+/**
+ * This file is part of Hyperf.
+ *
+ * @link     https://www.hyperf.io
+ * @document https://hyperf.wiki
+ * @contact  [email protected]
+ * @license  https://github.com/hyperf/hyperf/blob/master/LICENSE
+ */
+use PhpCsFixer\Config;
+use PhpCsFixer\Finder;
+
 $header = <<<'EOF'
 This file is part of Hyperf.
 
@@ -9,7 +21,7 @@
 @license  https://github.com/hyperf/hyperf/blob/master/LICENSE
 EOF;
 
-return (new PhpCsFixer\Config())
+return (new Config())
     ->setRiskyAllowed(true)
     ->setRules([
         '@PSR2' => true,
@@ -23,13 +35,13 @@
             'location' => 'after_declare_strict',
         ],
         'array_syntax' => [
-            'syntax' => 'short'
+            'syntax' => 'short',
         ],
         'list_syntax' => [
-            'syntax' => 'short'
+            'syntax' => 'short',
         ],
         'concat_space' => [
-            'spacing' => 'one'
+            'spacing' => 'one',
         ],
         'blank_line_before_statement' => [
             'statements' => [
@@ -38,7 +50,7 @@
         ],
         'general_phpdoc_annotation_remove' => [
             'annotations' => [
-                'author'
+                'author',
             ],
         ],
         'ordered_imports' => [
@@ -85,9 +97,10 @@
         'single_quote' => true,
         'standardize_not_equals' => true,
         'multiline_comment_opening_closing' => true,
+        'single_line_empty_body' => false,
     ])
     ->setFinder(
-        PhpCsFixer\Finder::create()
+        Finder::create()
             ->exclude('vendor')
             ->in(__DIR__)
     )

composer.json

@@ -22,7 +22,7 @@
         "ext-sockets": "*",
         "hyperf/coordinator": ">=3.0",
         "hyperf/coroutine": ">=3.0",
-        "multiplex/multiplex": "^1.0"
+        "multiplex/multiplex": "^1.1"
     },
     "require-dev": {
         "friendsofphp/php-cs-fixer": "^3.0",

src/Client.php

@@ -59,6 +59,7 @@ class Client implements ClientInterface, HasSerializerInterface
         'connect_timeout' => 0.5,
         'heartbeat' => 20,
         'max_requests' => 0,
+        'max_wait_close_seconds' => 2,
     ];
 
     protected ChannelManager $channelManager;
@@ -107,6 +108,7 @@ public function send(mixed $data): int
             $key = spl_object_hash($this);
             if (Locker::lock($key)) {
                 try {
+                    $this->waitUntilChannelManagerEmpty();
                     $this->close();
                 } finally {
                     Locker::unlock($key);
@@ -342,4 +344,23 @@ protected function loop(): void
             }
         });
     }
+
+    protected function waitUntilChannelManagerEmpty(): void
+    {
+        $now = microtime(true);
+        $seconds = $this->config['max_wait_close_seconds'];
+        while (true) {
+            if ($this->channelManager->isEmpty()) {
+                return;
+            }
+
+            if (microtime(true) - $now >= $seconds) {
+                return;
+            }
+
+            if (CoordinatorManager::until(Constants::WORKER_EXIT)->yield(0.05)) {
+                return;
+            }
+        }
+    }
 }

tests/Cases/ClientTest.php

@@ -28,6 +28,7 @@
 use Multiplex\Socket\Client;
 use Throwable;
 
+use function Hyperf\Coroutine\go;
 use function Hyperf\Coroutine\parallel;
 use function Hyperf\Tappable\tap;
 
@@ -282,4 +283,30 @@ public function testPushToClosedChannel()
             $this->assertFalse($chan->push(1));
         });
     }
+
+    public function testWaitUntilChannelManagerEmpty()
+    {
+        $this->runInCoroutine(function () {
+            $client = new Client('127.0.0.1', 9999);
+            $client->set([
+                'max_wait_close_seconds' => 0.5,
+            ]);
+            $ref = new ClassInvoker($client);
+            $manager = $client->getChannelManager();
+            $manager->get(1, true);
+            $now = microtime(true);
+            $ref->waitUntilChannelManagerEmpty();
+            $this->assertTrue(microtime(true) - $now > 0.4);
+            $this->assertTrue(microtime(true) - $now < 0.6);
+
+            $now = microtime(true);
+            go(function () use ($manager) {
+                usleep(1000 * 100);
+                $manager->close(1);
+            });
+            $ref->waitUntilChannelManagerEmpty();
+            $this->assertTrue(microtime(true) - $now > 0.08);
+            $this->assertTrue(microtime(true) - $now < 0.2);
+        });
+    }
 }