This commit is contained in:
2021-02-23 17:19:46 +08:00
parent 21bdf32fc3
commit 31d5024663
10 changed files with 41 additions and 26 deletions
+4 -2
View File
@@ -10,6 +10,7 @@ use RdKafka\ConsumerTopic;
use RdKafka\Exception;
use RdKafka\KafkaConsumer;
use RdKafka\TopicConf;
use Snowflake\Exception\ComponentException;
use Snowflake\Exception\ConfigException;
use Snowflake\Snowflake;
use Swoole\Coroutine;
@@ -98,6 +99,7 @@ class Kafka extends \Snowflake\Process\Process
/**
* @param $topic
* @param $message
* @throws ComponentException
*/
protected function handlerExecute($topic, $message)
{
@@ -109,11 +111,11 @@ class Kafka extends \Snowflake\Process\Process
if (!class_exists($namespace)) {
return;
}
$class = Snowflake::createObject($namespace);
$class = objectPool($namespace);
if (!($class instanceof ConsumerInterface)) {
return;
}
$class->onHandler(new Struct($topic, $message));
$class->onHandler(objectPool(Struct::class, [$topic, $message]));
} catch (Throwable $exception) {
$this->application->error($exception);
}