This commit is contained in:
as2252258@163.com
2021-04-24 20:44:38 +08:00
parent 263b09fd51
commit 2ffbea70c8
3 changed files with 12 additions and 6 deletions
+3 -1
View File
@@ -92,7 +92,9 @@ class OnPipeMessage extends Callback
{ {
[$topic, $rdMessage] = $message['body']; [$topic, $rdMessage] = $message['body'];
call_user_func($message['handler'], new Struct($topic, $rdMessage)); $message['handler'][0]->setParams(new Struct($topic, $rdMessage));
call_user_func($message['handler']);
return 'success'; return 'success';
} }
+3
View File
@@ -19,4 +19,7 @@ interface ConsumerInterface
public function onHandler(Struct $struct): void; public function onHandler(Struct $struct): void;
public function setParams(...$params): void;
} }
+1
View File
@@ -118,6 +118,7 @@ class Kafka extends \Snowflake\Process\Process
if (empty($handler)) { if (empty($handler)) {
return; return;
} }
$message = swoole_serialize(['action' => 'kafka', 'handler' => $handler, 'body' => [$topic, $message]]); $message = swoole_serialize(['action' => 'kafka', 'handler' => $handler, 'body' => [$topic, $message]]);
$server->sendMessage($message, random_int(0, $setting - 1)); $server->sendMessage($message, random_int(0, $setting - 1));