Skip to content

Commit 9cb853c

Browse files
makasimmakasim
committed
[consumption] Add OnStartContext.
1 parent ea74202 commit 9cb853c

16 files changed

Lines changed: 313 additions & 57 deletions

‎pkg/enqueue/Client/ConsumptionExtension/SetupBrokerExtension.php‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -3,9 +3,9 @@
33
namespace Enqueue\Client\ConsumptionExtension;
44

55
use Enqueue\Client\DriverInterface;
6-
use Enqueue\Consumption\Context;
76
use Enqueue\Consumption\EmptyExtensionTrait;
87
use Enqueue\Consumption\ExtensionInterface;
8+
use Enqueue\Consumption\OnStartContext;
99

1010
class SetupBrokerExtension implements ExtensionInterface
1111
{
@@ -33,7 +33,7 @@ public function __construct(DriverInterface $driver)
3333
/**
3434
* {@inheritdoc}
3535
*/
36-
public function onStart(Context $context)
36+
public function onStart(OnStartContext $context)
3737
{
3838
if (false == $this->isDone) {
3939
$this->isDone = true;

‎pkg/enqueue/Consumption/ChainExtension.php‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -20,9 +20,9 @@ public function __construct(array $extensions)
2020
}
2121

2222
/**
23-
* @param Context $context
23+
* {@inheritdoc}
2424
*/
25-
public function onStart(Context $context)
25+
public function onStart(OnStartContext $context)
2626
{
2727
foreach ($this->extensions as $extension) {
2828
$extension->onStart($context);

‎pkg/enqueue/Consumption/EmptyExtensionTrait.php‎

Lines changed: 8 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -5,49 +5,49 @@
55
trait EmptyExtensionTrait
66
{
77
/**
8-
* @param Context $context
8+
* {@inheritdoc}
99
*/
10-
public function onStart(Context $context)
10+
public function onStart(OnStartContext $context)
1111
{
1212
}
1313

1414
/**
15-
* @param Context $context
15+
* {@inheritdoc}
1616
*/
1717
public function onBeforeReceive(Context $context)
1818
{
1919
}
2020

2121
/**
22-
* @param Context $context
22+
* {@inheritdoc}
2323
*/
2424
public function onPreReceived(Context $context)
2525
{
2626
}
2727

2828
/**
29-
* @param Context $context
29+
* {@inheritdoc}
3030
*/
3131
public function onResult(Context $context)
3232
{
3333
}
3434

3535
/**
36-
* @param Context $context
36+
* {@inheritdoc}
3737
*/
3838
public function onPostReceived(Context $context)
3939
{
4040
}
4141

4242
/**
43-
* @param Context $context
43+
* {@inheritdoc}
4444
*/
4545
public function onIdle(Context $context)
4646
{
4747
}
4848

4949
/**
50-
* @param Context $context
50+
* {@inheritdoc}
5151
*/
5252
public function onInterrupted(Context $context)
5353
{

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

Lines changed: 20 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -5,9 +5,11 @@
55
use Enqueue\Consumption\Context;
66
use Enqueue\Consumption\EmptyExtensionTrait;
77
use Enqueue\Consumption\ExtensionInterface;
8+
use Enqueue\Consumption\OnStartContext;
89
use Enqueue\Consumption\Result;
910
use Interop\Queue\PsrMessage;
1011
use Psr\Log\LoggerInterface;
12+
use Psr\Log\NullLogger;
1113

1214
class LoggerExtension implements ExtensionInterface
1315
{
@@ -18,28 +20,38 @@ class LoggerExtension implements ExtensionInterface
1820
*/
1921
private $logger;
2022

23+
/**
24+
* @var bool
25+
*/
26+
private $replaceExisting;
27+
2128
/**
2229
* @param LoggerInterface $logger
30+
* @param bool $replaceExisting
2331
*/
24-
public function __construct(LoggerInterface $logger)
32+
public function __construct(LoggerInterface $logger, $replaceExisting = true)
2533
{
2634
$this->logger = $logger;
35+
$this->replaceExisting = $replaceExisting;
2736
}
2837

2938
/**
3039
* {@inheritdoc}
3140
*/
32-
public function onStart(Context $context)
41+
public function onStart(OnStartContext $context)
3342
{
34-
if ($context->getLogger()) {
43+
if ($context->getLogger() instanceof NullLogger) {
44+
$context->setLogger($this->logger);
45+
$this->logger->debug(sprintf('Set context\'s logger "%s"', get_class($this->logger)));
46+
} elseif ($this->replaceExisting) {
47+
$context->setLogger($this->logger);
3548
$context->getLogger()->debug(sprintf(
36-
'Skip setting context\'s logger "%s". Another one "%s" has already been set.',
37-
get_class($this->logger),
38-
get_class($context->getLogger())
49+
'Replace context\'s logger "%s" with "%s"',
50+
get_class($context->getLogger()),
51+
get_class($this->logger)
3952
));
4053
} else {
41-
$context->setLogger($this->logger);
42-
$this->logger->debug(sprintf('Set context\'s logger "%s"', get_class($this->logger)));
54+
$context->getLogger()->debug(sprintf('Skip setting a logger "%s"', get_class($this->logger)));
4355
}
4456
}
4557

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

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@
66
use Enqueue\Consumption\EmptyExtensionTrait;
77
use Enqueue\Consumption\Exception\LogicException;
88
use Enqueue\Consumption\ExtensionInterface;
9+
use Enqueue\Consumption\OnStartContext;
910
use Psr\Log\LoggerInterface;
1011

1112
class SignalExtension implements ExtensionInterface
@@ -25,7 +26,7 @@ class SignalExtension implements ExtensionInterface
2526
/**
2627
* {@inheritdoc}
2728
*/
28-
public function onStart(Context $context)
29+
public function onStart(OnStartContext $context)
2930
{
3031
if (false == extension_loaded('pcntl')) {
3132
throw new LogicException('The pcntl extension is required in order to catch signals.');

‎pkg/enqueue/Consumption/ExtensionInterface.php‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -8,9 +8,9 @@ interface ExtensionInterface
88
* Executed only once at the very begining of the consumption.
99
* At this stage the context does not contain processor, consumer and queue.
1010
*
11-
* @param Context $context
11+
* @param OnStartContext $context
1212
*/
13-
public function onStart(Context $context);
13+
public function onStart(OnStartContext $context);
1414

1515
/**
1616
* Executed at every new cycle before we asked a broker for a new message.

‎pkg/enqueue/Consumption/OnStartContext.php‎

Lines changed: 94 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,94 @@
1+
<?php
2+
namespace Enqueue\Consumption;
3+
4+
5+
use Interop\Queue\PsrConsumer;
6+
use Interop\Queue\PsrContext;
7+
use Interop\Queue\PsrProcessor;
8+
use Psr\Log\LoggerInterface;
9+
10+
class OnStartContext
11+
{
12+
/**
13+
* @var PsrContext
14+
*/
15+
private $psrContext;
16+
17+
/**
18+
* @var LoggerInterface
19+
*/
20+
private $logger;
21+
22+
/**
23+
* @var callable[]|PsrProcessor[]
24+
*/
25+
private $processors;
26+
/**
27+
* @var PsrConsumer[]
28+
*/
29+
private $consumers;
30+
31+
public function __construct(PsrContext $psrContext, LoggerInterface $logger, array $processors, array $consumers)
32+
{
33+
$this->psrContext = $psrContext;
34+
$this->logger = $logger;
35+
$this->processors = $processors;
36+
$this->consumers = $consumers;
37+
}
38+
39+
/**
40+
* @return callable[]|PsrProcessor[]
41+
*/
42+
public function getProcessors()
43+
{
44+
return $this->processors;
45+
}
46+
47+
/**
48+
* @param callable[]|PsrProcessor[] $processors
49+
*/
50+
public function setProcessors($processors)
51+
{
52+
$this->processors = $processors;
53+
}
54+
55+
/**
56+
* @return PsrConsumer[]
57+
*/
58+
public function getConsumers()
59+
{
60+
return $this->consumers;
61+
}
62+
63+
/**
64+
* @param PsrConsumer[] $consumers
65+
*/
66+
public function setConsumers(array $consumers)
67+
{
68+
$this->consumers = $consumers;
69+
}
70+
71+
/**
72+
* @return LoggerInterface
73+
*/
74+
public function getLogger()
75+
{
76+
return $this->logger;
77+
}
78+
79+
/**
80+
* @param LoggerInterface $logger
81+
*/
82+
public function setLogger(LoggerInterface $logger)
83+
{
84+
$this->logger = $logger;
85+
}
86+
87+
/**
88+
* @return PsrContext
89+
*/
90+
public function getPsrContext()
91+
{
92+
return $this->psrContext;
93+
}
94+
}

‎pkg/enqueue/Consumption/QueueConsumer.php‎

Lines changed: 12 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -170,15 +170,25 @@ public function consume(ExtensionInterface $runtimeExtension = null)
170170
$consumers[$queue->getQueueName()] = $this->psrContext->createConsumer($queue);
171171
}
172172

173+
$processors = [];
174+
/** @var PsrQueue $queue */
175+
foreach ($this->boundProcessors as list($queue, $processor)) {
176+
$processors[$queue->getQueueName()] = $processor;
177+
}
178+
173179
$this->extension = $runtimeExtension ?
174180
new ChainExtension([$this->staticExtension, $runtimeExtension]) :
175181
$this->staticExtension
176182
;
177183

178184
$context = new Context($this->psrContext);
179-
$this->extension->onStart($context);
185+
$onStartContext = new OnStartContext($this->psrContext, new NullLogger(), $processors, $consumers);
186+
$this->extension->onStart($onStartContext);
187+
188+
// BC
189+
$context->setLogger($onStartContext->getLogger());
180190

181-
$this->logger = $context->getLogger() ?: new NullLogger();
191+
$this->logger = $context->getLogger();
182192
$this->logger->info('Start consuming');
183193

184194
if ($this->psrContext instanceof AmqpContext) {

‎pkg/enqueue/Tests/Client/ConsumptionExtension/FlushSpoolProducerExtensionTest.php‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -7,11 +7,13 @@
77
use Enqueue\Consumption\Context;
88
use Enqueue\Consumption\ExtensionInterface;
99
use Enqueue\Test\ClassExtensionTrait;
10+
use Enqueue\Test\ConsumptionContextMockTrait;
1011
use PHPUnit\Framework\TestCase;
1112

1213
class FlushSpoolProducerExtensionTest extends TestCase
1314
{
1415
use ClassExtensionTrait;
16+
use ConsumptionContextMockTrait;
1517

1618
public function testShouldImplementExtensionInterface()
1719
{
@@ -32,7 +34,7 @@ public function testShouldDoNothingOnStart()
3234
;
3335

3436
$extension = new FlushSpoolProducerExtension($producer);
35-
$extension->onStart($this->createContextMock());
37+
$extension->onStart($this->createOnStartContextMock());
3638
}
3739

3840
public function testShouldDoNothingOnBeforeReceive()

‎pkg/enqueue/Tests/Client/ConsumptionExtension/SetupBrokerExtensionTest.php‎

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -4,16 +4,18 @@
44

55
use Enqueue\Client\ConsumptionExtension\SetupBrokerExtension;
66
use Enqueue\Client\DriverInterface;
7-
use Enqueue\Consumption\Context;
87
use Enqueue\Consumption\ExtensionInterface;
8+
use Enqueue\Consumption\OnStartContext;
99
use Enqueue\Test\ClassExtensionTrait;
10+
use Enqueue\Test\ConsumptionContextMockTrait;
1011
use Interop\Queue\PsrContext;
1112
use PHPUnit\Framework\TestCase;
1213
use Psr\Log\NullLogger;
1314

1415
class SetupBrokerExtensionTest extends TestCase
1516
{
1617
use ClassExtensionTrait;
18+
use ConsumptionContextMockTrait;
1719

1820
public function testShouldImplementExtensionInterface()
1921
{
@@ -36,8 +38,7 @@ public function testShouldSetupBroker()
3638
->with($this->identicalTo($logger))
3739
;
3840

39-
$context = new Context($this->createMock(PsrContext::class));
40-
$context->setLogger($logger);
41+
$context = new OnStartContext($this->createMock(PsrContext::class), $logger, [], []);
4142

4243
$extension = new SetupBrokerExtension($driver);
4344
$extension->onStart($context);
@@ -54,8 +55,7 @@ public function testShouldSetupBrokerOnlyOnce()
5455
->with($this->identicalTo($logger))
5556
;
5657

57-
$context = new Context($this->createMock(PsrContext::class));
58-
$context->setLogger($logger);
58+
$context = new OnStartContext($this->createMock(PsrContext::class), $logger, [], []);
5959

6060
$extension = new SetupBrokerExtension($driver);
6161
$extension->onStart($context);

0 commit comments

Comments
 (0)