Skip to content

Commit b018c84

Browse files
arrilotarrilot
authored
Merge pull request #4 from ensi-platform/task-83277
#83277
2 parents 6a25089 + 96cee6e commit b018c84

16 files changed

+211
-71
lines changed

‎.git_hooks/pre-commit/php-cs-fixer.sh‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,7 @@
22

33
EXECUTABLE_NAME=php-cs-fixer
44
EXECUTABLE_COMMAND=fix
5-
CONFIG_FILE=.php_cs
5+
CONFIG_FILE=.php-cs-fixer.php
66
CONFIG_FILE_PARAMETER='--config'
77
ROOT=`pwd`
88
ESC_SEQ="\x1b["

‎.gitignore‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,5 @@
11
.idea
2-
.php_cs
3-
.php_cs.cache
2+
.php-cs-fixer.cache
43
.phpunit.result.cache
54
build
65
composer.lock

‎.php-cs-fixer.php‎

Lines changed: 41 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,41 @@
1+
<?php
2+
3+
$finder = Symfony\Component\Finder\Finder::create()
4+
->in([
5+
__DIR__ . '/src',
6+
__DIR__ . '/tests',
7+
])
8+
->name('*.php')
9+
->notName('*.blade.php')
10+
->ignoreDotFiles(true)
11+
->ignoreVCS(true);
12+
13+
return (new PhpCsFixer\Config())
14+
->setRules([
15+
'@PSR2' => true,
16+
'array_syntax' => ['syntax' => 'short'],
17+
'ordered_imports' => ['sort_algorithm' => 'alpha'],
18+
'no_unused_imports' => true,
19+
'trailing_comma_in_multiline' => true,
20+
'phpdoc_scalar' => true,
21+
'unary_operator_spaces' => true,
22+
'blank_line_before_statement' => [
23+
'statements' => ['break', 'continue', 'declare', 'return', 'throw', 'try'],
24+
],
25+
'phpdoc_single_line_var_spacing' => true,
26+
'phpdoc_var_without_name' => true,
27+
'class_attributes_separation' => [
28+
'elements' => [
29+
'method' => 'one',
30+
],
31+
],
32+
'method_argument_space' => [
33+
'on_multiline' => 'ensure_fully_multiline',
34+
'keep_multiple_spaces_after_comma' => true,
35+
],
36+
'single_trait_insert_per_statement' => true,
37+
'no_whitespace_in_blank_line' => true,
38+
'method_chaining_indentation' => true,
39+
40+
])
41+
->setFinder($finder);

‎composer.json‎

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -11,21 +11,27 @@
1111
"authors": [
1212
{
1313
"name": "arrilot",
14-
"email": "nekrasov@ensi.ru",
14+
"email": "nekrasov@greensight.ru",
1515
"role": "Developer"
1616
}
1717
],
1818
"require": {
1919
"php": "^8.0",
2020
"ext-rdkafka": "*",
2121
"ensi/laravel-phprdkafka": "^0.2",
22-
"illuminate/support": "^7 || ^8"
22+
"illuminate/contracts": "^7.30 || ^8.37",
23+
"illuminate/pipeline": "^7.30 || ^8.37",
24+
"illuminate/support": "^7.30 || ^8.37"
2325
},
2426
"require-dev": {
2527
"brianium/paratest": "^6.2",
28+
"friendsofphp/php-cs-fixer": "^3.2",
2629
"nunomaduro/collision": "^5.3",
2730
"orchestra/testbench": "^6.15",
31+
"pestphp/pest": "^1.18",
32+
"pestphp/pest-plugin-laravel": "^1.1",
2833
"phpunit/phpunit": "^9.3",
34+
"php-parallel-lint/php-var-dump-check": "^0.5.0",
2935
"spatie/laravel-ray": "^1.9"
3036
},
3137
"autoload": {
@@ -39,6 +45,7 @@
3945
}
4046
},
4147
"scripts": {
48+
"cs": "php-cs-fixer fix --config .php-cs-fixer.php",
4249
"test": "./vendor/bin/testbench package:test --no-coverage",
4350
"test-coverage": "vendor/bin/phpunit --coverage-html coverage"
4451
},

‎config/kafka-consumer.php‎

Lines changed: 17 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,8 @@
11
<?php
22

33
return [
4+
'global_middleware' => [],
5+
46
'processors' => [
57
[
68
/*
@@ -41,12 +43,23 @@
4143
| - `<your-favorite-queue-name-as-string>` - stream message to this queue;
4244
*/
4345
'queue' => false,
46+
],
47+
],
48+
49+
'consumer_options' => [
50+
/** options for consumer with name `default` */
51+
'default' => [
52+
/*
53+
| Optional, defaults to 20000.
54+
| Kafka consume timeout in milliseconds.
55+
*/
56+
'consume_timeout' => 20000,
4457

4558
/*
46-
| Optional, defaults to 5000.
47-
| Kafka consume timeout in milliseconds .
59+
| Optional, defaults to empty array.
60+
| Array of middleware.
4861
*/
49-
'consume_timeout' => 5000,
62+
'middleware' => [],
5063
]
51-
],
64+
]
5265
];

‎src/Commands/KafkaConsumeCommand.php‎

Lines changed: 35 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -4,8 +4,9 @@
44

55
use Ensi\LaravelPhpRdKafkaConsumer\ConsumerOptions;
66
use Ensi\LaravelPhpRdKafkaConsumer\HighLevelConsumer;
7-
use Throwable;
7+
use Ensi\LaravelPhpRdKafkaConsumer\ProcessorData;
88
use Illuminate\Console\Command;
9+
use Throwable;
910

1011
class KafkaConsumeCommand extends Command
1112
{
@@ -28,7 +29,7 @@ class KafkaConsumeCommand extends Command
2829
/**
2930
* Execute the console command.
3031
*/
31-
public function handle(): int
32+
public function handle(HighLevelConsumer $highLevelConsumer): int
3233
{
3334
$topic = $this->argument('topic');
3435
$consumer = $this->argument('consumer');
@@ -49,34 +50,33 @@ public function handle(): int
4950
return 1;
5051
}
5152

52-
$processorClassName = $processorData['class'];
53-
if (!class_exists($processorClassName)) {
54-
$this->error("Processor class \"$processorClassName\" is not found");
53+
if (!class_exists($processorData->class)) {
54+
$this->error("Processor class \"$processorData->class\" is not found");
5555
$this->line('Processors are set in /config/kafka-consumers.php');
5656

5757
return 1;
5858
}
5959

60-
$supportedProcessorTypes = ['action', 'job'];
61-
$processorType = $processorData['type'] ?? 'action';
62-
if (!in_array($processorType, $supportedProcessorTypes)) {
63-
$this->error("Invalid processor type \"$processorType\", supported types are: " . implode(',', $supportedProcessorTypes));
60+
if (!$processorData->hasValidType()) {
61+
$this->error("Invalid processor type \"$processorData->type\", supported types are: " . implode(',', $processorData->getSupportedTypes()));
6462

6563
return 1;
6664
}
6765

68-
$processorQueue = $processorData['queue'] ?? false;
69-
66+
$consumerPackageOptions = config('kafka-consumer.consumer_options.'. $consumer, []);
7067
$consumerOptions = new ConsumerOptions(
71-
consumeTimeout: $processorData['consume_timeout'] ?? 20000,
68+
consumeTimeout: $consumerPackageOptions['consume_timeout'] ?? $processorData->consumeTimeout,
7269
maxEvents: $this->option('once') ? 1 : (int) $this->option('max-events'),
73-
maxTime: (int) $this->option('max-time')
70+
maxTime: (int) $this->option('max-time'),
71+
middleware: $this->collectMiddleware($consumerPackageOptions['middleware'] ?? []),
7472
);
7573

7674
$this->info("Start listenning to topic: \"$topic\", consumer \"$consumer\"");
75+
7776
try {
78-
$kafkaTopicListener = new HighLevelConsumer($topic, $consumer, $consumerOptions);
79-
$kafkaTopicListener->listen($processorClassName, $processorType, $processorQueue);
77+
$highLevelConsumer
78+
->for($consumer)
79+
->listen($topic, $processorData, $consumerOptions);
8080
} catch (Throwable $e) {
8181
$this->error('An error occurred while listening to the topic: '. $e->getMessage(). ' '. $e->getFile() . '::' . $e->getLine());
8282

@@ -86,17 +86,34 @@ public function handle(): int
8686
return 0;
8787
}
8888

89-
protected function findMatchedProcessor(string $topic, string $consumer): ?array
89+
protected function findMatchedProcessor(string $topic, string $consumer): ?ProcessorData
9090
{
9191
foreach (config('kafka-consumer.processors', []) as $processor) {
9292
if (
9393
(empty($processor['topic']) || $processor['topic'] === $topic)
9494
&& (empty($processor['consumer']) || $processor['consumer'] === $consumer)
9595
) {
96-
return $processor;
96+
return new ProcessorData(
97+
class: $processor['class'],
98+
topic: $processor['topic'] ?? null,
99+
consumer: $processor['consumer'] ?? null,
100+
type: $processor['type'] ?? 'action',
101+
queue: $processor['queue'] ?? false,
102+
consumeTimeout: $processor['consume_timeout'] ?? 20000,
103+
);
97104
}
98105
}
99106

100107
return null;
101108
}
102-
}
109+
110+
protected function collectMiddleware(array $processorMiddleware): array
111+
{
112+
return array_unique(
113+
array_merge(
114+
config('kafka-consumer.global_middleware', []),
115+
$processorMiddleware
116+
)
117+
);
118+
}
119+
}

‎src/ConsumerOptions.php‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@ public function __construct(
88
public int $consumeTimeout = 20000,
99
public int $maxEvents = 0,
1010
public int $maxTime = 0,
11-
)
12-
{
11+
public array $middleware = [],
12+
) {
1313
}
1414
}

‎src/Exceptions/KafkaConsumerException.php‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -6,5 +6,4 @@
66

77
class KafkaConsumerException extends Exception
88
{
9-
10-
}
9+
}

‎src/HighLevelConsumer.php‎

Lines changed: 44 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -4,45 +4,52 @@
44

55
use Ensi\LaravelPhpRdKafka\KafkaManager;
66
use Ensi\LaravelPhpRdKafkaConsumer\Exceptions\KafkaConsumerException;
7+
use Illuminate\Pipeline\Pipeline;
78
use RdKafka\Exception as RdKafkaException;
89
use RdKafka\KafkaConsumer;
910
use RdKafka\Message;
1011
use Throwable;
1112

1213
class HighLevelConsumer
1314
{
14-
protected KafkaConsumer $consumer;
15+
protected ?KafkaConsumer $consumer;
1516

1617
public function __construct(
17-
protected string $topicName,
18-
?string $consumerName = null,
19-
protected ConsumerOptions $options,
20-
)
18+
protected KafkaManager $kafkaManager,
19+
protected Pipeline $pipeline
20+
) {
21+
}
22+
23+
public function for(?string $consumerName): static
2124
{
22-
$manager = resolve(KafkaManager::class);
23-
$this->consumer = is_null($consumerName) ? $manager->consumer() : $manager->consumer($consumerName);
25+
$this->consumer = is_null($consumerName)
26+
? $this->kafkaManager->consumer()
27+
: $this->kafkaManager->consumer($consumerName);
28+
29+
return $this;
2430
}
2531

2632
/**
2733
* @throws KafkaException
2834
* @throws RdKafkaException
2935
* @throws Throwable
3036
*/
31-
public function listen(string $processorClassName, string $processorType, string|bool $processorQueue): void
37+
public function listen(string $topicName, ProcessorData $processorData, ConsumerOptions $options): void
3238
{
33-
$this->consumer->subscribe([ $this->topicName ]);
39+
$this->consumer->subscribe([ $topicName ]);
3440

3541
[$startTime, $eventsProcessed] = [hrtime(true) / 1e9, 0];
3642

3743
while (true) {
38-
$message = $this->consumer->consume($this->options->consumeTimeout);
44+
$message = $this->consumer->consume($options->consumeTimeout);
3945

4046
switch ($message->err) {
4147

4248
case RD_KAFKA_RESP_ERR_NO_ERROR:
43-
$this->executeProcessor($processorClassName, $processorType, $processorQueue, $message);
49+
$this->processThroughMiddleware($processorData, $message, $options);
4450
$this->consumer->commitAsync($message);
4551
$eventsProcessed++;
52+
4653
break;
4754

4855
case RD_KAFKA_RESP_ERR__TIMED_OUT:
@@ -55,45 +62,56 @@ public function listen(string $processorClassName, string $processorType, string
5562
throw new KafkaConsumerException('Kafka error: ' . $message->errstr());
5663
}
5764

58-
if ($this->shouldBeStopped($startTime, $eventsProcessed)) {
65+
if ($this->shouldBeStopped($startTime, $eventsProcessed, $options)) {
5966
break;
6067
}
6168
}
6269
}
6370

64-
protected function executeProcessor(string $className, string $type, string|bool $queue, Message $message): void
71+
protected function processThroughMiddleware(ProcessorData $processorData, Message $message, ConsumerOptions $options): void
6572
{
66-
$queue
67-
? $this->executeQueueableProcessor($className, $type, $queue, $message)
68-
: $this->executeSyncProcessor($className, $type, $message);
73+
$this->pipeline
74+
->send($message)
75+
->through($options->middleware)
76+
->then(fn (Message $message) => $this->executeProcessor($processorData, $message));
6977
}
7078

71-
protected function executeSyncProcessor(string $className, string $type, Message $message): void
79+
protected function executeProcessor(ProcessorData $processorData, Message $message): void
7280
{
73-
if ($type === 'job') {
81+
$processorData->queue
82+
? $this->executeQueueableProcessor($processorData, $message)
83+
: $this->executeSyncProcessor($processorData, $message);
84+
}
85+
86+
protected function executeSyncProcessor(ProcessorData $processorData, Message $message): void
87+
{
88+
$className = $processorData->class;
89+
if ($processorData->type === 'job') {
7490
$className::dispatchSync($message);
75-
} elseif ($type === 'action') {
91+
} elseif ($processorData->type === 'action') {
7692
resolve($className)->execute($message);
7793
}
7894
}
7995

80-
protected function executeQueueableProcessor(string $className, string $type, string|bool $queue, Message $message): void
96+
protected function executeQueueableProcessor(ProcessorData $processorData, Message $message): void
8197
{
82-
if ($type === 'job') {
98+
$className = $processorData->class;
99+
$queue = $processorData->queue;
100+
if ($processorData->type === 'job') {
83101
is_string($queue) ? $className::dispatch($message)->onQueue($queue) : $className::dispatch($message);
84-
} elseif ($type === 'action') {
102+
} elseif ($processorData->type === 'action') {
85103
$processor = resolve($className);
86104
is_string($queue) ? $processor->onQueue($queue)->execute($message) : $processor->execute($message);
87105
}
88106
}
89107

90-
protected function shouldBeStopped(int|float $startTime, int $eventsProcessed): bool
108+
protected function shouldBeStopped(int|float $startTime, int $eventsProcessed, ConsumerOptions $options): bool
91109
{
92-
if ($this->options->maxTime && hrtime(true) / 1e9 - $startTime >= $this->options->maxTime) {
110+
if ($options->maxTime && hrtime(true) / 1e9 - $startTime >= $options->maxTime) {
93111
return true;
94-
}
112+
}
95113

96-
if ($this->options->maxEvents && $eventsProcessed >= $this->options->maxEvents) {
114+
if ($options->maxEvents && $eventsProcessed >= $options->maxEvents) {
97115
return true;
98116
}
99117

0 commit comments

Comments
 (0)