[Messenger] Notify transports which messages are still being processed, using pcntl_alarm()

Author: HypeMC

CHANGELOG.md

@@ -1,6 +1,11 @@
 CHANGELOG
 =========
 
+7.2
+---
+
+ * Implement the `KeepaliveReceiverInterface` to enable asynchronously notifying Beanstalkd that the job is still being processed, in order to avoid timeouts
+
 5.2.0
 -----
 

Tests/Transport/BeanstalkdReceiverTest.php

@@ -16,6 +16,7 @@
 use Symfony\Component\Messenger\Bridge\Beanstalkd\Transport\BeanstalkdReceivedStamp;
 use Symfony\Component\Messenger\Bridge\Beanstalkd\Transport\BeanstalkdReceiver;
 use Symfony\Component\Messenger\Bridge\Beanstalkd\Transport\Connection;
+use Symfony\Component\Messenger\Envelope;
 use Symfony\Component\Messenger\Exception\MessageDecodingFailedException;
 use Symfony\Component\Messenger\Transport\Serialization\PhpSerializer;
 use Symfony\Component\Messenger\Transport\Serialization\Serializer;
@@ -78,6 +79,17 @@ public function testItRejectTheMessageIfThereIsAMessageDecodingFailedException()
         $receiver->get();
     }
 
+    public function testKeepalive()
+    {
+        $serializer = $this->createSerializer();
+
+        $connection = $this->createMock(Connection::class);
+        $connection->expects($this->once())->method('keepalive')->with(1);
+
+        $receiver = new BeanstalkdReceiver($connection, $serializer);
+        $receiver->keepalive(new Envelope(new DummyMessage('foo'), [new BeanstalkdReceivedStamp(1, 'bar')]));
+    }
+
     private function createBeanstalkdEnvelope(): array
     {
         return [

Tests/Transport/BeanstalkdTransportTest.php

@@ -13,6 +13,7 @@
 
 use PHPUnit\Framework\TestCase;
 use Symfony\Component\Messenger\Bridge\Beanstalkd\Tests\Fixtures\DummyMessage;
+use Symfony\Component\Messenger\Bridge\Beanstalkd\Transport\BeanstalkdReceivedStamp;
 use Symfony\Component\Messenger\Bridge\Beanstalkd\Transport\BeanstalkdTransport;
 use Symfony\Component\Messenger\Bridge\Beanstalkd\Transport\Connection;
 use Symfony\Component\Messenger\Envelope;
@@ -50,6 +51,18 @@ public function testReceivesMessages()
         $this->assertSame($decodedMessage, $envelopes[0]->getMessage());
     }
 
+    public function testKeepalive()
+    {
+        $transport = $this->getTransport(
+            null,
+            $connection = $this->createMock(Connection::class),
+        );
+
+        $connection->expects($this->once())->method('keepalive')->with(1);
+
+        $transport->keepalive(new Envelope(new DummyMessage('foo'), [new BeanstalkdReceivedStamp(1, 'bar')]));
+    }
+
     private function getTransport(?SerializerInterface $serializer = null, ?Connection $connection = null): BeanstalkdTransport
     {
         $serializer ??= $this->createMock(SerializerInterface::class);

Tests/Transport/ConnectionTest.php

@@ -330,4 +330,37 @@ public function testSendWhenABeanstalkdExceptionOccurs()
 
         $connection->send($body, $headers, $delay);
     }
+
+    public function testKeepalive()
+    {
+        $id = 123456;
+
+        $tube = 'baz';
+
+        $client = $this->createMock(PheanstalkInterface::class);
+        $client->expects($this->once())->method('useTube')->with($tube)->willReturn($client);
+        $client->expects($this->once())->method('touch')->with($this->callback(fn (JobId $jobId): bool => $jobId->getId() === $id));
+
+        $connection = new Connection(['tube_name' => $tube], $client);
+
+        $connection->keepalive((string) $id);
+    }
+
+    public function testKeepaliveWhenABeanstalkdExceptionOccurs()
+    {
+        $id = 123456;
+
+        $tube = 'baz123';
+
+        $exception = new ServerException('baz error');
+
+        $client = $this->createMock(PheanstalkInterface::class);
+        $client->expects($this->once())->method('useTube')->with($tube)->willReturn($client);
+        $client->expects($this->once())->method('touch')->with($this->callback(fn (JobId $jobId): bool => $jobId->getId() === $id))->willThrowException($exception);
+
+        $connection = new Connection(['tube_name' => $tube], $client);
+
+        $this->expectExceptionObject(new TransportException($exception->getMessage(), 0, $exception));
+        $connection->keepalive((string) $id);
+    }
 }

Transport/BeanstalkdReceiver.php

@@ -14,15 +14,15 @@
 use Symfony\Component\Messenger\Envelope;
 use Symfony\Component\Messenger\Exception\LogicException;
 use Symfony\Component\Messenger\Exception\MessageDecodingFailedException;
+use Symfony\Component\Messenger\Transport\Receiver\KeepaliveReceiverInterface;
 use Symfony\Component\Messenger\Transport\Receiver\MessageCountAwareInterface;
-use Symfony\Component\Messenger\Transport\Receiver\ReceiverInterface;
 use Symfony\Component\Messenger\Transport\Serialization\PhpSerializer;
 use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;
 
 /**
  * @author Antonio Pauletich <[email protected]>
  */
-class BeanstalkdReceiver implements ReceiverInterface, MessageCountAwareInterface
+class BeanstalkdReceiver implements KeepaliveReceiverInterface, MessageCountAwareInterface
 {
     private SerializerInterface $serializer;
 
@@ -65,6 +65,11 @@ public function reject(Envelope $envelope): void
         $this->connection->reject($this->findBeanstalkdReceivedStamp($envelope)->getId());
     }
 
+    public function keepalive(Envelope $envelope): void
+    {
+        $this->connection->keepalive($this->findBeanstalkdReceivedStamp($envelope)->getId());
+    }
+
     public function getMessageCount(): int
     {
         return $this->connection->getMessageCount();

Transport/BeanstalkdTransport.php

@@ -12,6 +12,7 @@
 namespace Symfony\Component\Messenger\Bridge\Beanstalkd\Transport;
 
 use Symfony\Component\Messenger\Envelope;
+use Symfony\Component\Messenger\Transport\Receiver\KeepaliveReceiverInterface;
 use Symfony\Component\Messenger\Transport\Receiver\MessageCountAwareInterface;
 use Symfony\Component\Messenger\Transport\Serialization\PhpSerializer;
 use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;
@@ -20,7 +21,7 @@
 /**
  * @author Antonio Pauletich <[email protected]>
  */
-class BeanstalkdTransport implements TransportInterface, MessageCountAwareInterface
+class BeanstalkdTransport implements TransportInterface, KeepaliveReceiverInterface, MessageCountAwareInterface
 {
     private SerializerInterface $serializer;
     private BeanstalkdReceiver $receiver;
@@ -48,6 +49,11 @@ public function reject(Envelope $envelope): void
         $this->getReceiver()->reject($envelope);
     }
 
+    public function keepalive(Envelope $envelope): void
+    {
+        $this->getReceiver()->keepalive($envelope);
+    }
+
     public function getMessageCount(): int
     {
         return $this->getReceiver()->getMessageCount();

Transport/Connection.php

@@ -180,6 +180,15 @@ public function reject(string $id): void
         }
     }
 
+    public function keepalive(string $id): void
+    {
+        try {
+            $this->client->useTube($this->tube)->touch(new JobId((int) $id));
+        } catch (Exception $exception) {
+            throw new TransportException($exception->getMessage(), 0, $exception);
+        }
+    }
+
     public function getMessageCount(): int
     {
         try {

composer.json

@@ -14,7 +14,7 @@
     "require": {
         "php": ">=8.2",
         "pda/pheanstalk": "^4.0",
-        "symfony/messenger": "^6.4|^7.0"
+        "symfony/messenger": "^7.2"
     },
     "require-dev": {
         "symfony/property-access": "^6.4|^7.0",