改名
This commit is contained in:
+4
-5
@@ -38,6 +38,9 @@ class Kafka extends \Snowflake\Process\Process
|
|||||||
$config->setGroupId($kafka['groupId']);
|
$config->setGroupId($kafka['groupId']);
|
||||||
$config->setBrokerVersion($kafka['version']);
|
$config->setBrokerVersion($kafka['version']);
|
||||||
$config->setTopics($kafka['topics']);
|
$config->setTopics($kafka['topics']);
|
||||||
|
$this->channelListener();
|
||||||
|
|
||||||
|
return [new Consumer(), $kafka];
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
@@ -47,11 +50,7 @@ class Kafka extends \Snowflake\Process\Process
|
|||||||
*/
|
*/
|
||||||
public function onHandler(Process $process)
|
public function onHandler(Process $process)
|
||||||
{
|
{
|
||||||
$this->initConfig();
|
[$consumer, $kafka] = $this->initConfig();
|
||||||
$this->channelListener();
|
|
||||||
|
|
||||||
$consumer = new Consumer();
|
|
||||||
$kafka = SConfig::get('kafka');
|
|
||||||
if ($kafka['debug'] ?? true) {
|
if ($kafka['debug'] ?? true) {
|
||||||
$consumer->setLogger(new Logger());
|
$consumer->setLogger(new Logger());
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user