Skip to content

Commit 10e05d6

Browse files
makasimmakasim
committed
[consumption] add InitLogger extension point. Before start.
1 parent 98dd6b5 commit 10e05d6

13 files changed

+182
-260
lines changed

‎pkg/enqueue/Consumption/ChainExtension.php

Lines changed: 17 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
namespace Enqueue\Consumption;
44

55
use Enqueue\Consumption\Context\End;
6+
use Enqueue\Consumption\Context\InitLogger;
67
use Enqueue\Consumption\Context\MessageReceived;
78
use Enqueue\Consumption\Context\MessageResult;
89
use Enqueue\Consumption\Context\PostConsume;
@@ -15,6 +16,7 @@
1516
class ChainExtension implements ExtensionInterface
1617
{
1718
private $startExtensions;
19+
private $initLoggerExtensions;
1820
private $preSubscribeExtensions;
1921
private $preConsumeExtensions;
2022
private $messageReceivedExtensions;
@@ -24,12 +26,10 @@ class ChainExtension implements ExtensionInterface
2426
private $postConsumeExtensions;
2527
private $endExtensions;
2628

27-
/**
28-
* @param ExtensionInterface[] $extensions
29-
*/
3029
public function __construct(array $extensions)
3130
{
3231
$this->startExtensions = [];
32+
$this->initLoggerExtensions = [];
3333
$this->preSubscribeExtensions = [];
3434
$this->preConsumeExtensions = [];
3535
$this->messageReceivedExtensions = [];
@@ -42,6 +42,7 @@ public function __construct(array $extensions)
4242
array_walk($extensions, function ($extension) {
4343
if ($extension instanceof ExtensionInterface) {
4444
$this->startExtensions[] = $extension;
45+
$this->initLoggerExtensions[] = $extension;
4546
$this->preSubscribeExtensions[] = $extension;
4647
$this->preConsumeExtensions[] = $extension;
4748
$this->messageReceivedExtensions[] = $extension;
@@ -61,6 +62,12 @@ public function __construct(array $extensions)
6162
$extensionValid = true;
6263
}
6364

65+
if ($extension instanceof InitLoggerExtensionInterface) {
66+
$this->initLoggerExtensions[] = $extension;
67+
68+
$extensionValid = true;
69+
}
70+
6471
if ($extension instanceof PreSubscribeExtensionInterface) {
6572
$this->preSubscribeExtensions[] = $extension;
6673

@@ -115,6 +122,13 @@ public function __construct(array $extensions)
115122
});
116123
}
117124

125+
public function onInitLogger(InitLogger $context): void
126+
{
127+
foreach ($this->endExtensions as $extension) {
128+
$extension->onInitLogger($context);
129+
}
130+
}
131+
118132
public function onStart(Start $context): void
119133
{
120134
foreach ($this->startExtensions as $extension) {

‎pkg/enqueue/Consumption/Context/InitLogger.php

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,28 @@
1+
<?php
2+
3+
namespace Enqueue\Consumption\Context;
4+
5+
use Psr\Log\LoggerInterface;
6+
7+
class InitLogger
8+
{
9+
/**
10+
* @var LoggerInterface
11+
*/
12+
private $logger;
13+
14+
public function __construct(LoggerInterface $logger)
15+
{
16+
$this->logger = $logger;
17+
}
18+
19+
public function getLogger(): LoggerInterface
20+
{
21+
return $this->logger;
22+
}
23+
24+
public function changeLogger(LoggerInterface $logger): void
25+
{
26+
$this->logger = $logger;
27+
}
28+
}

‎pkg/enqueue/Consumption/Context/Start.php

Lines changed: 0 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -62,11 +62,6 @@ public function getLogger(): LoggerInterface
6262
return $this->logger;
6363
}
6464

65-
public function changeLogger(LoggerInterface $logger): void
66-
{
67-
$this->logger = $logger;
68-
}
69-
7065
/**
7166
* In milliseconds.
7267
*/

‎pkg/enqueue/Consumption/Extension/LoggerExtension.php

Lines changed: 8 additions & 58 deletions
Original file line numberDiff line numberDiff line change
@@ -2,16 +2,11 @@
22

33
namespace Enqueue\Consumption\Extension;
44

5-
use Enqueue\Consumption\Context\PostMessageReceived;
6-
use Enqueue\Consumption\Context\Start;
7-
use Enqueue\Consumption\PostMessageReceivedExtensionInterface;
8-
use Enqueue\Consumption\Result;
9-
use Enqueue\Consumption\StartExtensionInterface;
10-
use Interop\Queue\Message as InteropMessage;
5+
use Enqueue\Consumption\Context\InitLogger;
6+
use Enqueue\Consumption\InitLoggerExtensionInterface;
117
use Psr\Log\LoggerInterface;
12-
use Psr\Log\NullLogger;
138

14-
class LoggerExtension implements StartExtensionInterface, PostMessageReceivedExtensionInterface
9+
class LoggerExtension implements InitLoggerExtensionInterface
1510
{
1611
/**
1712
* @var LoggerInterface
@@ -26,59 +21,14 @@ public function __construct(LoggerInterface $logger)
2621
$this->logger = $logger;
2722
}
2823

29-
public function onStart(Start $context): void
24+
public function onInitLogger(InitLogger $context): void
3025
{
31-
if ($context->getLogger() && false == $context->getLogger() instanceof NullLogger) {
32-
$context->getLogger()->debug(sprintf(
33-
'Skip setting context\'s logger "%s". Another one "%s" has already been set.',
34-
get_class($this->logger),
35-
get_class($context->getLogger())
36-
));
37-
} else {
38-
$context->changeLogger($this->logger);
39-
$this->logger->debug(sprintf('Set context\'s logger "%s"', get_class($this->logger)));
40-
}
41-
}
42-
43-
public function onPostMessageReceived(PostMessageReceived $context): void
44-
{
45-
if (false == $context->getResult() instanceof Result) {
46-
return;
47-
}
48-
49-
/** @var $result Result */
50-
$result = $context->getResult();
26+
$previousLogger = $context->getLogger();
5127

52-
switch ($result->getStatus()) {
53-
case Result::REJECT:
54-
case Result::REQUEUE:
55-
if ($result->getReason()) {
56-
$this->logger->error($result->getReason(), $this->messageToLogContext($context->getMessage()));
57-
}
58-
59-
break;
60-
case Result::ACK:
61-
if ($result->getReason()) {
62-
$this->logger->info($result->getReason(), $this->messageToLogContext($context->getMessage()));
63-
}
28+
if ($previousLogger !== $this->logger) {
29+
$context->changeLogger($this->logger);
6430

65-
break;
66-
default:
67-
throw new \LogicException(sprintf('Got unexpected message result. "%s"', $result->getStatus()));
31+
$this->logger->debug(sprintf('Change logger from "%s" to "%s"', get_class($previousLogger), get_class($this->logger)));
6832
}
6933
}
70-
71-
/**
72-
* @param InteropMessage $message
73-
*
74-
* @return array
75-
*/
76-
private function messageToLogContext(InteropMessage $message)
77-
{
78-
return [
79-
'body' => $message->getBody(),
80-
'headers' => $message->getHeaders(),
81-
'properties' => $message->getProperties(),
82-
];
83-
}
8434
}

‎pkg/enqueue/Consumption/ExtensionInterface.php

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

33
namespace Enqueue\Consumption;
44

5-
interface ExtensionInterface extends StartExtensionInterface, PreSubscribeExtensionInterface, PreConsumeExtensionInterface, MessageReceivedExtensionInterface, PostMessageReceivedExtensionInterface, MessageResultExtensionInterface, ProcessorExceptionExtensionInterface, PostConsumeExtensionInterface, EndExtensionInterface
5+
interface ExtensionInterface extends StartExtensionInterface, PreSubscribeExtensionInterface, PreConsumeExtensionInterface, MessageReceivedExtensionInterface, PostMessageReceivedExtensionInterface, MessageResultExtensionInterface, ProcessorExceptionExtensionInterface, PostConsumeExtensionInterface, EndExtensionInterface, InitLoggerExtensionInterface
66
{
77
}

‎pkg/enqueue/Consumption/InitLoggerExtensionInterface.php

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,14 @@
1+
<?php
2+
3+
namespace Enqueue\Consumption;
4+
5+
use Enqueue\Consumption\Context\InitLogger;
6+
7+
interface InitLoggerExtensionInterface
8+
{
9+
/**
10+
* Executed only once at the very beginning of the QueueConsumer::consume method call.
11+
* BEFORE onStart extension method.
12+
*/
13+
public function onInitLogger(InitLogger $context): void;
14+
}

‎pkg/enqueue/Consumption/QueueConsumer.php

Lines changed: 24 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
namespace Enqueue\Consumption;
44

55
use Enqueue\Consumption\Context\End;
6+
use Enqueue\Consumption\Context\InitLogger;
67
use Enqueue\Consumption\Context\MessageReceived;
78
use Enqueue\Consumption\Context\MessageResult;
89
use Enqueue\Consumption\Context\PostConsume;
@@ -32,7 +33,7 @@ final class QueueConsumer implements QueueConsumerInterface
3233
private $interopContext;
3334

3435
/**
35-
* @var ExtensionInterface|ChainExtension
36+
* @var ExtensionInterface
3637
*/
3738
private $staticExtension;
3839

@@ -46,11 +47,6 @@ final class QueueConsumer implements QueueConsumerInterface
4647
*/
4748
private $receiveTimeout;
4849

49-
/**
50-
* @var ExtensionInterface|ChainExtension
51-
*/
52-
private $extension;
53-
5450
/**
5551
* @var LoggerInterface
5652
*/
@@ -128,11 +124,16 @@ public function bindCallback($queue, callable $processor): QueueConsumerInterfac
128124

129125
public function consume(ExtensionInterface $runtimeExtension = null): void
130126
{
131-
$this->extension = $runtimeExtension ?
127+
$extension = $runtimeExtension ?
132128
new ChainExtension([$this->staticExtension, $runtimeExtension]) :
133129
$this->staticExtension
134130
;
135131

132+
$initLogger = new InitLogger($this->logger);
133+
$extension->onInitLogger($initLogger);
134+
135+
$this->logger = $initLogger->getLogger();
136+
136137
$startTime = (int) (microtime(true) * 1000);
137138

138139
$start = new Start(
@@ -143,10 +144,10 @@ public function consume(ExtensionInterface $runtimeExtension = null): void
143144
$startTime
144145
);
145146

146-
$this->extension->onStart($start);
147+
$extension->onStart($start);
147148

148149
if ($start->isExecutionInterrupted()) {
149-
$this->onEnd($startTime);
150+
$this->onEnd($extension, $startTime);
150151

151152
return;
152153
}
@@ -176,7 +177,7 @@ public function consume(ExtensionInterface $runtimeExtension = null): void
176177
$receivedMessagesCount = 0;
177178
$interruptExecution = false;
178179

179-
$callback = function (InteropMessage $message, Consumer $consumer) use (&$receivedMessagesCount, &$interruptExecution) {
180+
$callback = function (InteropMessage $message, Consumer $consumer) use (&$receivedMessagesCount, &$interruptExecution, $extension) {
180181
++$receivedMessagesCount;
181182

182183
$receivedAt = (int) (microtime(true) * 1000);
@@ -188,19 +189,19 @@ public function consume(ExtensionInterface $runtimeExtension = null): void
188189
$processor = $this->boundProcessors[$queue->getQueueName()]->getProcessor();
189190

190191
$messageReceived = new MessageReceived($this->interopContext, $consumer, $message, $processor, $receivedAt, $this->logger);
191-
$this->extension->onMessageReceived($messageReceived);
192+
$extension->onMessageReceived($messageReceived);
192193
$result = $messageReceived->getResult();
193194
$processor = $messageReceived->getProcessor();
194195
if (null === $result) {
195196
try {
196197
$result = $processor->process($message, $this->interopContext);
197198
} catch (\Exception $e) {
198-
$result = $this->onProcessorException($message, $e, $receivedAt);
199+
$result = $this->onProcessorException($extension, $message, $e, $receivedAt);
199200
}
200201
}
201202

202203
$messageResult = new MessageResult($this->interopContext, $message, $result, $receivedAt, $this->logger);
203-
$this->extension->onResult($messageResult);
204+
$extension->onResult($messageResult);
204205
$result = $messageResult->getResult();
205206

206207
switch ($result) {
@@ -220,7 +221,7 @@ public function consume(ExtensionInterface $runtimeExtension = null): void
220221
}
221222

222223
$postMessageReceived = new PostMessageReceived($this->interopContext, $message, $result, $receivedAt, $this->logger);
223-
$this->extension->onPostMessageReceived($postMessageReceived);
224+
$extension->onPostMessageReceived($postMessageReceived);
224225

225226
if ($postMessageReceived->isExecutionInterrupted()) {
226227
$interruptExecution = true;
@@ -241,7 +242,7 @@ public function consume(ExtensionInterface $runtimeExtension = null): void
241242
$this->logger
242243
);
243244

244-
$this->extension->onPreSubscribe($preSubscribe);
245+
$extension->onPreSubscribe($preSubscribe);
245246

246247
$subscriptionConsumer->subscribe($consumer, $callback);
247248
}
@@ -252,21 +253,21 @@ public function consume(ExtensionInterface $runtimeExtension = null): void
252253
$interruptExecution = false;
253254

254255
$preConsume = new PreConsume($this->interopContext, $subscriptionConsumer, $this->logger, $cycle, $this->receiveTimeout, $startTime);
255-
$this->extension->onPreConsume($preConsume);
256+
$extension->onPreConsume($preConsume);
256257

257258
if ($preConsume->isExecutionInterrupted()) {
258-
$this->onEnd($startTime, $subscriptionConsumer);
259+
$this->onEnd($extension, $startTime, $subscriptionConsumer);
259260

260261
return;
261262
}
262263

263264
$subscriptionConsumer->consume($this->receiveTimeout);
264265

265266
$postConsume = new PostConsume($this->interopContext, $subscriptionConsumer, $receivedMessagesCount, $cycle, $startTime, $this->logger);
266-
$this->extension->onPostConsume($postConsume);
267+
$extension->onPostConsume($postConsume);
267268

268269
if ($interruptExecution || $postConsume->isExecutionInterrupted()) {
269-
$this->onEnd($startTime, $subscriptionConsumer);
270+
$this->onEnd($extension, $startTime, $subscriptionConsumer);
270271

271272
return;
272273
}
@@ -285,11 +286,11 @@ public function setFallbackSubscriptionConsumer(SubscriptionConsumer $fallbackSu
285286
$this->fallbackSubscriptionConsumer = $fallbackSubscriptionConsumer;
286287
}
287288

288-
private function onEnd(int $startTime, SubscriptionConsumer $subscriptionConsumer = null): void
289+
private function onEnd(ExtensionInterface $extension, int $startTime, SubscriptionConsumer $subscriptionConsumer = null): void
289290
{
290291
$endTime = (int) (microtime(true) * 1000);
291292

292-
$this->extension->onEnd(new End($this->interopContext, $startTime, $endTime, $this->logger));
293+
$extension->onEnd(new End($this->interopContext, $startTime, $endTime, $this->logger));
293294

294295
if ($subscriptionConsumer) {
295296
$subscriptionConsumer->unsubscribeAll();
@@ -301,12 +302,12 @@ private function onEnd(int $startTime, SubscriptionConsumer $subscriptionConsume
301302
*
302303
* https://github.com/symfony/symfony/blob/cbe289517470eeea27162fd2d523eb29c95f775f/src/Symfony/Component/HttpKernel/EventListener/ExceptionListener.php#L77
303304
*/
304-
private function onProcessorException(Message $message, \Exception $exception, int $receivedAt)
305+
private function onProcessorException(ExtensionInterface $extension, Message $message, \Exception $exception, int $receivedAt)
305306
{
306307
$processorException = new ProcessorException($this->interopContext, $message, $exception, $receivedAt, $this->logger);
307308

308309
try {
309-
$this->extension->onProcessorException($processorException);
310+
$extension->onProcessorException($processorException);
310311

311312
$result = $processorException->getResult();
312313
if (null === $result) {

‎pkg/enqueue/Consumption/QueueConsumerInterface.php

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -35,7 +35,7 @@ public function bindCallback($queue, callable $processor): self;
3535
* Runtime extension - is an extension or a collection of extensions which could be set on runtime.
3636
* Here's a good example: @see LimitsExtensionsCommandTrait.
3737
*
38-
* @param ExtensionInterface|ChainExtension|null $runtimeExtension
38+
* @param ExtensionInterface|null $runtimeExtension
3939
*
4040
* @throws \Exception
4141
*/

‎pkg/enqueue/Symfony/Consumption/LimitsExtensionsCommandTrait.php

Lines changed: 0 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -13,9 +13,6 @@
1313

1414
trait LimitsExtensionsCommandTrait
1515
{
16-
/**
17-
* {@inheritdoc}
18-
*/
1916
protected function configureLimitsExtensions()
2017
{
2118
$this

0 commit comments

Comments
 (0)