Skip to content

Commit 4c5b60a

Browse files
makasimmakasim
authored
Merge pull request #46 from php-enqueue/consumption-on-result-extension-point
[consumption] Add onResult extension point.
2 parents e75617e + 94ddbda commit 4c5b60a

File tree

7 files changed

+184
-3
lines changed

7 files changed

+184
-3
lines changed

‎pkg/enqueue/Consumption/ChainExtension.php‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -49,6 +49,16 @@ public function onPreReceived(Context $context)
4949
}
5050
}
5151

52+
/**
53+
* @param Context $context
54+
*/
55+
public function onResult(Context $context)
56+
{
57+
foreach ($this->extensions as $extension) {
58+
$extension->onResult($context);
59+
}
60+
}
61+
5262
/**
5363
* @param Context $context
5464
*/

‎pkg/enqueue/Consumption/EmptyExtensionTrait.php‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,13 @@ public function onPreReceived(Context $context)
2525
{
2626
}
2727

28+
/**
29+
* @param Context $context
30+
*/
31+
public function onResult(Context $context)
32+
{
33+
}
34+
2835
/**
2936
* @param Context $context
3037
*/

‎pkg/enqueue/Consumption/ExtensionInterface.php‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,15 @@ public function onBeforeReceive(Context $context);
3131
*/
3232
public function onPreReceived(Context $context);
3333

34+
/**
35+
* Executed when a message is processed by a processor or a result was set in onPreReceived method.
36+
* BUT before the message status was sent to the broker
37+
* The consumption could be interrupted at this step but it exits after the message is processed.
38+
*
39+
* @param Context $context
40+
*/
41+
public function onResult(Context $context);
42+
3443
/**
3544
* Executed when a message is processed by a processor.
3645
* The context contains a status, which could not be changed.

‎pkg/enqueue/Consumption/QueueConsumer.php‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -204,6 +204,8 @@ protected function doConsume(ExtensionInterface $extension, Context $context)
204204
$context->setResult($result);
205205
}
206206

207+
$extension->onResult($context);
208+
207209
switch ($context->getResult()) {
208210
case Result::ACK:
209211
$consumer->acknowledge($message);

pkg/enqueue/Tests/Consumption/ExtensionsTest.php renamed to pkg/enqueue/Tests/Consumption/ChainExtensionTest.php

Lines changed: 23 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,7 @@
77
use Enqueue\Consumption\ExtensionInterface;
88
use Enqueue\Test\ClassExtensionTrait;
99

10-
class ExtensionsTest extends \PHPUnit_Framework_TestCase
10+
class ChainExtensionTest extends \PHPUnit_Framework_TestCase
1111
{
1212
use ClassExtensionTrait;
1313

@@ -87,6 +87,28 @@ public function testShouldProxyOnPreReceiveToAllInternalExtensions()
8787
$extensions->onPreReceived($context);
8888
}
8989

90+
public function testShouldProxyOnResultToAllInternalExtensions()
91+
{
92+
$context = $this->createContextMock();
93+
94+
$fooExtension = $this->createExtension();
95+
$fooExtension
96+
->expects($this->once())
97+
->method('onResult')
98+
->with($this->identicalTo($context))
99+
;
100+
$barExtension = $this->createExtension();
101+
$barExtension
102+
->expects($this->once())
103+
->method('onResult')
104+
->with($this->identicalTo($context))
105+
;
106+
107+
$extensions = new ChainExtension([$fooExtension, $barExtension]);
108+
109+
$extensions->onResult($context);
110+
}
111+
90112
public function testShouldProxyOnPostReceiveToAllInternalExtensions()
91113
{
92114
$context = $this->createContextMock();

‎pkg/enqueue/Tests/Consumption/EmptyExtensionTraitTest.php‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,19 @@
1+
<?php
2+
namespace Enqueue\Tests\Consumption;
3+
4+
use Enqueue\Consumption\EmptyExtensionTrait;
5+
use Enqueue\Consumption\ExtensionInterface;
6+
use PHPUnit\Framework\TestCase;
7+
8+
class EmptyExtensionTraitTest extends TestCase
9+
{
10+
public function testTraitMustImplementOrExtensionMethods()
11+
{
12+
new EmptyExtension();
13+
}
14+
}
15+
16+
class EmptyExtension implements ExtensionInterface
17+
{
18+
use EmptyExtensionTrait;
19+
}

‎pkg/enqueue/Tests/Consumption/QueueConsumerTest.php‎

Lines changed: 114 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -467,7 +467,7 @@ public function testShouldCallOnBeforeReceiveExtensionMethod()
467467
$queueConsumer->consume();
468468
}
469469

470-
public function testShouldCallOnPreReceivedAndPostReceivedExtensionMethods()
470+
public function testShouldCallOnPreReceivedExtensionMethodWithExpectedContext()
471471
{
472472
$expectedMessage = $this->createMessageMock();
473473
$messageConsumerStub = $this->createMessageConsumerStub($expectedMessage);
@@ -497,6 +497,62 @@ public function testShouldCallOnPreReceivedAndPostReceivedExtensionMethods()
497497
$this->assertFalse($context->isExecutionInterrupted());
498498
})
499499
;
500+
501+
$chainExtensions = new ChainExtension([$extension, new BreakCycleExtension(1)]);
502+
$queueConsumer = new QueueConsumer($contextStub, $chainExtensions, 0);
503+
$queueConsumer->bind(new NullQueue('aQueueName'), $processorMock);
504+
505+
$queueConsumer->consume();
506+
}
507+
508+
public function testShouldCallOnResultExtensionMethodWithExpectedContext()
509+
{
510+
$expectedMessage = $this->createMessageMock();
511+
$messageConsumerStub = $this->createMessageConsumerStub($expectedMessage);
512+
513+
$contextStub = $this->createPsrContextStub($messageConsumerStub);
514+
515+
$processorMock = $this->createProcessorStub();
516+
517+
$extension = $this->createExtension();
518+
$extension
519+
->expects($this->once())
520+
->method('onResult')
521+
->with($this->isInstanceOf(Context::class))
522+
->willReturnCallback(function (Context $context) use (
523+
$contextStub,
524+
$messageConsumerStub,
525+
$processorMock,
526+
$expectedMessage
527+
) {
528+
$this->assertSame($contextStub, $context->getPsrContext());
529+
$this->assertSame($messageConsumerStub, $context->getPsrConsumer());
530+
$this->assertSame($processorMock, $context->getPsrProcessor());
531+
$this->assertSame($expectedMessage, $context->getPsrMessage());
532+
$this->assertInstanceOf(NullLogger::class, $context->getLogger());
533+
$this->assertNull($context->getException());
534+
$this->assertSame(Result::ACK, $context->getResult());
535+
$this->assertFalse($context->isExecutionInterrupted());
536+
})
537+
;
538+
539+
$chainExtensions = new ChainExtension([$extension, new BreakCycleExtension(1)]);
540+
$queueConsumer = new QueueConsumer($contextStub, $chainExtensions, 0);
541+
$queueConsumer->bind(new NullQueue('aQueueName'), $processorMock);
542+
543+
$queueConsumer->consume();
544+
}
545+
546+
public function testShouldCallOnPostReceivedExtensionMethodWithExpectedContext()
547+
{
548+
$expectedMessage = $this->createMessageMock();
549+
$messageConsumerStub = $this->createMessageConsumerStub($expectedMessage);
550+
551+
$contextStub = $this->createPsrContextStub($messageConsumerStub);
552+
553+
$processorMock = $this->createProcessorStub();
554+
555+
$extension = $this->createExtension();
500556
$extension
501557
->expects($this->once())
502558
->method('onPostReceived')
@@ -722,6 +778,57 @@ public function testShouldAllowInterruptConsumingOnPreReceiveButProcessCurrentMe
722778
$queueConsumer->consume();
723779
}
724780

781+
public function testShouldAllowInterruptConsumingOnResult()
782+
{
783+
$expectedMessage = $this->createMessageMock();
784+
$messageConsumerStub = $this->createMessageConsumerStub($expectedMessage);
785+
786+
$contextStub = $this->createPsrContextStub($messageConsumerStub);
787+
788+
$processorMock = $this->createProcessorMock();
789+
$processorMock
790+
->expects($this->once())
791+
->method('process')
792+
->willReturn(Result::ACK)
793+
;
794+
795+
$extension = $this->createExtension();
796+
$extension
797+
->expects($this->once())
798+
->method('onResult')
799+
->with($this->isInstanceOf(Context::class))
800+
->willReturnCallback(function (Context $context) {
801+
$context->setExecutionInterrupted(true);
802+
})
803+
;
804+
$extension
805+
->expects($this->atLeastOnce())
806+
->method('onInterrupted')
807+
->with($this->isInstanceOf(Context::class))
808+
->willReturnCallback(function (Context $context) use (
809+
$contextStub,
810+
$messageConsumerStub,
811+
$processorMock,
812+
$expectedMessage
813+
) {
814+
$this->assertSame($contextStub, $context->getPsrContext());
815+
$this->assertSame($messageConsumerStub, $context->getPsrConsumer());
816+
$this->assertSame($processorMock, $context->getPsrProcessor());
817+
$this->assertSame($expectedMessage, $context->getPsrMessage());
818+
$this->assertInstanceOf(NullLogger::class, $context->getLogger());
819+
$this->assertNull($context->getException());
820+
$this->assertSame(Result::ACK, $context->getResult());
821+
$this->assertTrue($context->isExecutionInterrupted());
822+
})
823+
;
824+
825+
$chainExtensions = new ChainExtension([$extension, new BreakCycleExtension(1)]);
826+
$queueConsumer = new QueueConsumer($contextStub, $chainExtensions, 0);
827+
$queueConsumer->bind(new NullQueue('aQueueName'), $processorMock);
828+
829+
$queueConsumer->consume();
830+
}
831+
725832
public function testShouldAllowInterruptConsumingOnPostReceive()
726833
{
727834
$expectedMessage = $this->createMessageMock();
@@ -850,6 +957,11 @@ public function testShouldCallExtensionPassedOnRuntime()
850957
->method('onPreReceived')
851958
->with($this->isInstanceOf(Context::class))
852959
;
960+
$runtimeExtension
961+
->expects($this->once())
962+
->method('onResult')
963+
->with($this->isInstanceOf(Context::class))
964+
;
853965
$runtimeExtension
854966
->expects($this->once())
855967
->method('onPostReceived')
@@ -936,7 +1048,7 @@ public function testShouldCallEachQueueOneByOne()
9361048
})
9371049
;
9381050
$extension
939-
->expects($this->at(4))
1051+
->expects($this->at(5))
9401052
->method('onBeforeReceive')
9411053
->with($this->isInstanceOf(Context::class))
9421054
->willReturnCallback(function (Context $context) use ($anotherProcessorMock, $queue2) {

0 commit comments

Comments
 (0)