Skip to content

Commit 7141e01

Browse files
srozesroze
authored
Merge pull request #2 from continuouspipe/improve-the-workers
Improve the workers
2 parents a76a16f + f89636e commit 7141e01

8 files changed

Lines changed: 162 additions & 16 deletions

File tree

‎composer.json‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@
44
"psr-4": { "": "src/" }
55
},
66
"require": {
7-
"google/cloud": "^0.20.2"
7+
"google/cloud": "^0.20.2",
8+
"tolerance/tolerance": "dev-master"
89
}
910
}

‎src/ContinuousPipe/Message/AutoRetry/CatchDoctrineDriverException.php‎

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,25 @@
1+
<?php
2+
3+
namespace ContinuousPipe\Message\AutoRetry;
4+
5+
use Doctrine\DBAL\Exception\DriverException;
6+
use Tolerance\Operation\ExceptionCatcher\ThrowableCatcherVoter;
7+
8+
class CatchDoctrineDriverException implements ThrowableCatcherVoter
9+
{
10+
/**
11+
* {@inheritdoc}
12+
*/
13+
public function shouldCatchThrowable($throwable)
14+
{
15+
return $throwable instanceof DriverException;
16+
}
17+
18+
/**
19+
* {@inheritdoc}
20+
*/
21+
public function shouldCatch(\Exception $e)
22+
{
23+
return $this->shouldCatchThrowable($e);
24+
}
25+
}

‎src/ContinuousPipe/Message/GooglePubSub/PubSubMessagePuller.php‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@
33
namespace ContinuousPipe\Message\GooglePubSub;
44

55
use ContinuousPipe\Message\Message;
6+
use ContinuousPipe\Message\MessageDeadlineExpirationManager;
67
use ContinuousPipe\Message\MessageException;
78
use ContinuousPipe\Message\MessagePuller;
89
use Google\Cloud\Exception\GoogleException;
@@ -13,7 +14,7 @@
1314
use JMS\Serializer\SerializerInterface;
1415
use Psr\Log\LoggerInterface;
1516

16-
class PubSubMessagePuller implements MessagePuller
17+
class PubSubMessagePuller implements MessagePuller, MessageDeadlineExpirationManager
1718
{
1819
/**
1920
* @var ServiceBuilder

‎src/ContinuousPipe/Message/MessageDeadlineExpirationManager.php‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,8 @@
1+
<?php
2+
3+
namespace ContinuousPipe\Message;
4+
5+
interface MessageDeadlineExpirationManager
6+
{
7+
public function extendDeadline(string $messageIdentifier, int $seconds);
8+
}

‎src/ContinuousPipe/MessageBundle/Command/ExtendMessageDeadlineCommand.php‎

Lines changed: 28 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -2,33 +2,57 @@
22

33
namespace ContinuousPipe\MessageBundle\Command;
44

5+
use ContinuousPipe\Message\MessageDeadlineExpirationManager;
6+
use ContinuousPipe\Message\MessagePuller;
57
use Symfony\Bundle\FrameworkBundle\Command\ContainerAwareCommand;
68
use Symfony\Component\Console\Input\InputArgument;
79
use Symfony\Component\Console\Input\InputInterface;
810
use Symfony\Component\Console\Output\OutputInterface;
911

1012
class ExtendMessageDeadlineCommand extends ContainerAwareCommand
1113
{
14+
/**
15+
* @var MessageDeadlineExpirationManager
16+
*/
17+
private $messageDeadlineExpirationManager;
18+
19+
/**
20+
* @var bool
21+
*/
22+
private $shouldStop = false;
23+
24+
public function __construct(MessageDeadlineExpirationManager $messageDeadlineExpirationManager)
25+
{
26+
parent::__construct('continuouspipe:message:extend-deadline');
27+
28+
$this->messageDeadlineExpirationManager = $messageDeadlineExpirationManager;
29+
}
30+
1231
public function configure()
1332
{
1433
$this
15-
->setName('continuouspipe:message:extend-deadline')
1634
->addArgument('message-id', InputArgument::REQUIRED)
1735
;
1836
}
1937

2038
public function run(InputInterface $input, OutputInterface $output)
2139
{
22-
$poller = $this->getContainer()->get('continuouspipe.message.google_pub_sub.message_poller');
40+
pcntl_signal(SIGTERM, [$this, 'stopCommand']);
41+
pcntl_signal(SIGINT, [$this, 'stopCommand']);
2342

2443
$messageIdentifier = $input->getArgument('message-id');
2544
$expirationExtension = 60;
2645

27-
while (true) {
46+
while (!$this->shouldStop) {
2847
$output->write(sprintf('Extending the deadline of message "%s" by %d seconds', $messageIdentifier, $expirationExtension));
29-
$poller->extendDeadline($messageIdentifier, $expirationExtension);
48+
$this->messageDeadlineExpirationManager->extendDeadline($messageIdentifier, $expirationExtension);
3049

3150
sleep($expirationExtension - 5);
3251
}
3352
}
53+
54+
public function stopCommand()
55+
{
56+
$this->shouldStop = true;
57+
}
3458
}

‎src/ContinuousPipe/MessageBundle/Command/PullAndConsumeMessageCommand.php‎

Lines changed: 71 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -2,37 +2,77 @@
22

33
namespace ContinuousPipe\MessageBundle\Command;
44

5+
use ContinuousPipe\Message\MessageConsumer;
6+
use ContinuousPipe\Message\MessagePuller;
57
use ContinuousPipe\Message\PulledMessage;
68
use Doctrine\Common\Collections\ArrayCollection;
79
use Doctrine\Common\Collections\Collection;
10+
use Psr\Log\LoggerInterface;
811
use Symfony\Bundle\FrameworkBundle\Command\ContainerAwareCommand;
912
use Symfony\Component\Console\Input\InputInterface;
1013
use Symfony\Component\Console\Output\Output;
1114
use Symfony\Component\Console\Output\OutputInterface;
1215
use Symfony\Component\Process\Process;
16+
use Tolerance\Operation\ExceptionCatcher\ThrowableCatcherVoter;
1317

1418
class PullAndConsumeMessageCommand extends ContainerAwareCommand
1519
{
20+
/**
21+
* Maximum runtime, in seconds. 30 minutes, in order to prevent any database-timeout related issue.
22+
*/
23+
const MAX_RUNTIME_IN_SECS = 1800;
24+
25+
/**
26+
* @var bool
27+
*/
1628
private $shouldStop = false;
1729

18-
public function configure()
19-
{
20-
$this->setName('continuouspipe:message:pull-and-consume');
30+
/**
31+
* @var MessagePuller
32+
*/
33+
private $messagePuller;
34+
35+
/**
36+
* @var MessageConsumer
37+
*/
38+
private $messageConsumer;
39+
40+
/**
41+
* @var ThrowableCatcherVoter
42+
*/
43+
private $throwableCatcherVoter;
44+
45+
/**
46+
* @var LoggerInterface
47+
*/
48+
private $logger;
49+
50+
public function __construct(
51+
MessagePuller $messagePuller,
52+
MessageConsumer $messageConsumer,
53+
ThrowableCatcherVoter $throwableCatcherVoter,
54+
LoggerInterface $logger
55+
) {
56+
parent::__construct('continuouspipe:message:pull-and-consume');
57+
58+
$this->messagePuller = $messagePuller;
59+
$this->messageConsumer = $messageConsumer;
60+
$this->throwableCatcherVoter = $throwableCatcherVoter;
61+
$this->logger = $logger;
2162
}
2263

2364
public function run(InputInterface $input, OutputInterface $output)
2465
{
25-
$puller = $this->getContainer()->get('continuouspipe.message.message_poller');
26-
$consumer = $this->getContainer()->get('continuouspipe.message.message_consumer');
2766
$consolePath = $this->getContainer()->getParameter('kernel.root_dir').DIRECTORY_SEPARATOR.'console';
2867

2968
pcntl_signal(SIGTERM, [$this, 'stopCommand']);
3069
pcntl_signal(SIGINT, [$this, 'stopCommand']);
3170

3271
$output->writeln('Waiting for messages...');
72+
$startTime = time();
3373

3474
while (!$this->shouldStop) {
35-
foreach ($puller->pull() as $pulledMessage) {
75+
foreach ($this->messagePuller->pull() as $pulledMessage) {
3676
/** @var PulledMessage $pulledMessage */
3777
$message = $pulledMessage->getMessage();
3878

@@ -41,13 +81,34 @@ public function run(InputInterface $input, OutputInterface $output)
4181
$extenderProcess = new Process($consolePath . ' continuouspipe:message:extend-deadline ' . $pulledMessage->getAcknowledgeIdentifier());
4282
$extenderProcess->start();
4383

44-
$consumer->consume($message);
84+
try {
85+
$this->messageConsumer->consume($message);
86+
} catch (\Throwable $e) {
87+
// The throwable will be handled later...
88+
$this->logger->warning('An exception occurred while processing the message', [
89+
'exception' => $e,
90+
]);
91+
} finally {
92+
$extenderProcess->stop(0);
93+
}
94+
95+
if (isset($e) && $this->throwableCatcherVoter->shouldCatchThrowable($e)) {
96+
$output->writeln(sprintf('Message "%s" (%s) has not been acknowledge as the exception has been cought', get_class($message), $pulledMessage->getIdentifier()));
97+
} else {
98+
$output->writeln(sprintf('Acknowledging message "%s" (%s)', get_class($message), $pulledMessage->getIdentifier()));
99+
$pulledMessage->acknowledge();
100+
}
45101

46-
$output->writeln(sprintf('Acknowledging message "%s" (%s)', get_class($message), $pulledMessage->getIdentifier()));
47-
$pulledMessage->acknowledge();
48-
$extenderProcess->stop(0);
49102
$output->writeln(sprintf('Finished consuming message "%s" (%s)', get_class($message), $pulledMessage->getIdentifier()));
103+
104+
// If an exception happened, break the loop to ensure to go back to the `shouldStop` condition
105+
if (isset($e)) {
106+
break;
107+
}
50108
}
109+
110+
$ranMoreThanRunTime = (time() - $startTime) > self::MAX_RUNTIME_IN_SECS;
111+
$this->shouldStop = $this->shouldStop || $ranMoreThanRunTime;
51112
}
52113

53114
$output->writeln('The worker has stopped (should have stopped: '.($this->shouldStop ? 'yes' : 'no').')');

‎src/ContinuousPipe/MessageBundle/DependencyInjection/MessageExtension.php‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@ public function load(array $configs, ContainerBuilder $container)
1919

2020
$loader = new Loader\XmlFileLoader($container, new FileLocator(__DIR__.'/../Resources/config'));
2121
$loader->load('simple-bus.xml');
22+
$loader->load('command.xml');
2223
$loader->load('drivers/'.$config['driver'].'.xml');
2324
}
2425
}

‎src/ContinuousPipe/MessageBundle/Resources/config/command.xml‎

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,25 @@
1+
<?xml version="1.0" ?>
2+
3+
<container xmlns="http://symfony.com/schema/dic/services"
4+
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
5+
xsi:schemaLocation="http://symfony.com/schema/dic/services http://symfony.com/schema/dic/services/services-1.0.xsd">
6+
7+
<services>
8+
<service id="continuouspipe.message.command.pull_and_consumer.throwable_catcher.doctrine" class="ContinuousPipe\Message\AutoRetry\CatchDoctrineDriverException" />
9+
<service id="continuouspipe.message.command.pull_and_consumer.throwable_catcher" alias="continuouspipe.message.command.pull_and_consumer.throwable_catcher.doctrine" />
10+
<service id="continuouspipe.message.command.pull_and_consumer" class="ContinuousPipe\MessageBundle\Command\PullAndConsumeMessageCommand">
11+
<argument type="service" id="continuouspipe.message.message_poller" />
12+
<argument type="service" id="continuouspipe.message.message_consumer" />
13+
<argument type="service" id="continuouspipe.message.command.pull_and_consumer.throwable_catcher" />
14+
<argument type="service" id="logger" />
15+
16+
<tag name="console.command" />
17+
</service>
18+
19+
<service id="continuouspipe.message.command.extend_message_deadline" class="ContinuousPipe\MessageBundle\Command\ExtendMessageDeadlineCommand">
20+
<argument type="service" id="continuouspipe.message.google_pub_sub.message_poller"/>
21+
22+
<tag name="console.command" />
23+
</service>
24+
</services>
25+
</container>

0 commit comments

Comments
 (0)