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
2020-10-26 17:25:24 +08:00
parent a6274261a9
commit babf8f9691
2 changed files with 7 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Kafka/Kafka.php
+6 -6
View File
@@ -65,7 +65,7 @@ class Kafka extends \Snowflake\Process\Process
} }
switch ($message->err) { switch ($message->err) {
case RD_KAFKA_RESP_ERR_NO_ERROR: case RD_KAFKA_RESP_ERR_NO_ERROR:
$this->channel->push([$message->topic_name, $message->partition, $message]); $this->channel->push([$message->topic_name, $message]);
break; break;
case RD_KAFKA_RESP_ERR__PARTITION_EOF: case RD_KAFKA_RESP_ERR__PARTITION_EOF:
$this->application->error('No more messages; will wait for more'); $this->application->error('No more messages; will wait for more');
@@ -95,7 +95,7 @@ class Kafka extends \Snowflake\Process\Process
$group->done(); $group->done();
}); });
while ($messages = $this->channel->pop()) { while ($messages = $this->channel->pop()) {
$this->handlerExecute($messages[0], $messages[1], $messages[2]); $this->handlerExecute($messages[0], $messages[1]);
} }
}); });
} }
@@ -109,7 +109,7 @@ class Kafka extends \Snowflake\Process\Process
* @param $part * @param $part
* @param $message * @param $message
*/ */
protected function handlerExecute($topic, $part, $message) protected function handlerExecute($topic, $message)
{ {
try { try {
$namespace = 'App\\Kafka\\' . ucfirst($topic) . 'Consumer'; $namespace = 'App\\Kafka\\' . ucfirst($topic) . 'Consumer';
@@ -120,7 +120,7 @@ class Kafka extends \Snowflake\Process\Process
if (!($class instanceof ConsumerInterface)) { if (!($class instanceof ConsumerInterface)) {
return; return;
} }
$class->onHandler(new Struct($topic, $part, $message)); $class->onHandler(new Struct($topic, $message));
} catch (\Throwable $exception) { } catch (\Throwable $exception) {
$this->application->error($exception->getMessage()); $this->application->error($exception->getMessage());
} }
@@ -140,8 +140,8 @@ class Kafka extends \Snowflake\Process\Process
$conf->set('group.id', uniqid('kafka')); $conf->set('group.id', uniqid('kafka'));
$conf->set('group.id', 'myConsumerGroup'); $conf->set('group.id', 'myConsumerGroup');
$conf->set('metadata.broker.list', '127.0.0.1:9092'); // $conf->set('metadata.broker.list', '127.0.0.1:9092');
// $conf->set('metadata.broker.list', $kafka['brokers']); $conf->set('metadata.broker.list', $kafka['brokers']);
$conf->set('auto.offset.reset', 'earliest'); $conf->set('auto.offset.reset', 'earliest');
$conf->set('socket.timeout.ms', 300000); $conf->set('socket.timeout.ms', 300000);
Kafka/Struct.php
+1 -2
View File
@@ -20,10 +20,9 @@ class Struct
/** /**
* Struct constructor. * Struct constructor.
* @param $topic * @param $topic
* @param $part
* @param $message * @param $message
*/ */
public function __construct($topic, $part,Message $message) public function __construct($topic, Message $message)
{ {
$this->topic = $topic; $this->topic = $topic;
$this->offset = $message->offset; $this->offset = $message->offset;