as2252258/kiri-core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 2 Wiki Activity

改名

Browse Source
This commit is contained in:
Mr·x
2021-02-20 18:33:17 +08:00
parent ab57da86c0
commit a1f9f57fd2
1 changed files with 2 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Kafka/Kafka.php
+2 -2
View File
@@ -39,7 +39,7 @@ class Kafka extends \Snowflake\Process\Process
*/ */
public function onHandler(Process $process): void public function onHandler(Process $process): void
{ {
$this->channelListener(); // $this->channelListener();
$this->waite(json_decode($process->read(), true)); $this->waite(json_decode($process->read(), true));
Timer::tick(1000, function () { Timer::tick(1000, function () {
@@ -86,7 +86,7 @@ class Kafka extends \Snowflake\Process\Process
return; return;
} }
if ($message->err == RD_KAFKA_RESP_ERR_NO_ERROR) { if ($message->err == RD_KAFKA_RESP_ERR_NO_ERROR) {
$this->channel->push([$message->topic_name, $message]); $this->handlerExecute($message->topic_name, $message);
} else if ($message->err == RD_KAFKA_RESP_ERR__PARTITION_EOF) { } else if ($message->err == RD_KAFKA_RESP_ERR__PARTITION_EOF) {
$this->application->warning('No more messages; will wait for more'); $this->application->warning('No more messages; will wait for more');
} else if ($message->err == RD_KAFKA_RESP_ERR__TIMED_OUT) { } else if ($message->err == RD_KAFKA_RESP_ERR__TIMED_OUT) {