This commit is contained in:
2020-10-09 18:33:45 +08:00
parent b3b5d2f6e6
commit c7f6f810cd
+5 -4
View File
@@ -15,6 +15,7 @@
namespace Kafka\Consumer; namespace Kafka\Consumer;
use Swoole\Timer; use Swoole\Timer;
use function Amp\repeat;
/** /**
* +------------------------------------------------------------------------------ * +------------------------------------------------------------------------------
@@ -136,13 +137,13 @@ class State
continue; continue;
} }
$interval = isset($option['interval']) ? $option['interval'] : 200; $interval = isset($option['interval']) ? $option['interval'] : 200;
Timer::tick($msInterval = $interval, function ($watcherId) use ($request, $option) { repeat(function ($watcherId) use ($request, $option) {
if ($this->checkRun($request) && $option['func'] != null) { if ($this->checkRun($request) && $option['func'] != null) {
$context = call_user_func($option['func']); $context = call_user_func($option['func']);
$this->processing($request, $context); $this->processing($request, $context);
} }
$this->requests[$request]['watcher'] = $watcherId; $this->requests[$request]['watcher'] = $watcherId;
}); }, $msInterval = $interval);
} }
// start sync metadata // start sync metadata
@@ -150,9 +151,9 @@ class State
$context = call_user_func($this->requests[self::REQUEST_METADATA]['func']); $context = call_user_func($this->requests[self::REQUEST_METADATA]['func']);
$this->processing($request, $context); $this->processing($request, $context);
} }
Timer::tick($msInterval = 1000, function ($watcherId) { repeat(function ($watcherId) {
$this->report(); $this->report();
}); }, $msInterval = 1000);
} }
// }}} // }}}