From 8b1751eea71f4a81cefcb7b3e625afd8dc450d79 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E5=A6=82=E6=9E=9C=E7=9A=84=E5=A6=82=E6=9E=9C?=
 <admin@fosuss.com>
Date: Sun, 3 Jan 2021 21:18:13 +0800
Subject: [PATCH] 1

---
 src/CoroutineRunner/Runner.php | 29 +++++++++++++++++++++++++----
 1 file changed, 25 insertions(+), 4 deletions(-)

diff --git a/src/CoroutineRunner/Runner.php b/src/CoroutineRunner/Runner.php
index 8d6d85a..b1c8065 100644
--- a/src/CoroutineRunner/Runner.php
+++ b/src/CoroutineRunner/Runner.php
@@ -14,6 +14,8 @@ class Runner
     protected $isRunning = false;
     protected $runningNum = 0;
     protected $onException;
+    /** @var callable|null */
+    protected $onLoop;
 
     function __construct($concurrency = 64,$taskChannelSize = 1024)
     {
@@ -21,16 +23,22 @@ function __construct($concurrency = 64,$taskChannelSize = 1024)
         $this->taskChannel = new Channel($taskChannelSize);
     }
 
-    function setOnException(callable $call)
+    function setOnException(callable $call):Runner
     {
         $this->onException = $call;
         return $this;
     }
 
-    function status()
+    function setOnLoop(callable $call):Runner
+    {
+        $this->onLoop = $call;
+        return $this;
+    }
+
+    function status():array
     {
         return [
-            'taskNum'=>$this->taskChannel->stats(),
+            'queueSize'=>$this->taskChannel->length(),
             'concurrency'=>$this->concurrency,
             'runningNum'=>$this->runningNum,
             'isRunning'=>$this->isRunning
@@ -43,14 +51,23 @@ function addTask(Task $task):Runner
         return $this;
     }
 
+    function queueSize():int
+    {
+        return $this->taskChannel->length();
+    }
+
     function start(float $waitTime = 30)
     {
         if(!$this->isRunning){
             $this->isRunning = true;
+            $this->runningNum = 0;
+        }
+        if($waitTime <=0){
+            $waitTime = PHP_INT_MAX;
         }
         $start = time();
         while ($waitTime > 0){
-            if($this->runningNum < $this->concurrency && !$this->taskChannel->isEmpty()){
+            if($this->runningNum <= $this->concurrency && !$this->taskChannel->isEmpty()){
                 $task = $this->taskChannel->pop(0.01);
                 if($task instanceof Task){
                     Coroutine::create(function ()use($task){
@@ -88,7 +105,11 @@ function start(float $waitTime = 30)
                     Coroutine::sleep(0.01);
                 }
             }
+            if(is_callable($this->onLoop)){
+                call_user_func($this->onLoop,$this);
+            }
         }
         $this->isRunning = false;
+        $this->runningNum = 0;
     }
 }
\ No newline at end of file