symfony · fabpot · Oct 14, 2024 · Oct 13, 2024
@@ -139,6 +139,14 @@ public function setAlarmInterval(?int $seconds): void
         $this->scheduleAlarm();
     }
 
+    /**
+     * Gets the interval in seconds on which a SIGALRM signal is dispatched.
+     */
+    public function getAlarmInterval(): ?int
+    {
+        return $this->alarmInterval;
+    }
+
     private function scheduleAlarm(): void
     {
         if (null !== $this->alarmInterval) {

@@ -2305,6 +2305,7 @@ public function testAlarmDispatchWithoutEventDispatcher()
         $application = $this->createSignalableApplication($command, null);
 
         $this->assertSame(1, $application->run(new ArrayInput(['alarm'])));
+        $this->assertSame(1, $application->getAlarmInterval());
         $this->assertTrue($command->signaled);
     }
 

diff --git a/src/Symfony/Component/Messenger/Bridge/Beanstalkd/Transport/BeanstalkdReceiver.php b/src/Symfony/Component/Messenger/Bridge/Beanstalkd/Transport/BeanstalkdReceiver.php
@@ -65,7 +65,7 @@ public function reject(Envelope $envelope): void
         $this->connection->reject($this->findBeanstalkdReceivedStamp($envelope)->getId());
     }
 
-    public function keepalive(Envelope $envelope): void
+    public function keepalive(Envelope $envelope, ?int $seconds = null): void
     {
         $this->connection->keepalive($this->findBeanstalkdReceivedStamp($envelope)->getId());
     }

diff --git a/src/Symfony/Component/Messenger/Bridge/Beanstalkd/Transport/BeanstalkdTransport.php b/src/Symfony/Component/Messenger/Bridge/Beanstalkd/Transport/BeanstalkdTransport.php
@@ -49,9 +49,9 @@ public function reject(Envelope $envelope): void
         $this->getReceiver()->reject($envelope);
     }
 
-    public function keepalive(Envelope $envelope): void
+    public function keepalive(Envelope $envelope, ?int $seconds = null): void
     {
-        $this->getReceiver()->keepalive($envelope);
+        $this->getReceiver()->keepalive($envelope, $seconds);
     }
 
     public function getMessageCount(): int

diff --git a/src/Symfony/Component/Messenger/Command/ConsumeMessagesCommand.php b/src/Symfony/Component/Messenger/Command/ConsumeMessagesCommand.php
@@ -286,7 +286,7 @@ public function handleSignal(int $signal, int|false $previousExitCode = 0): int|
         if (\SIGALRM === $signal) {
             $this->logger?->info('Sending keepalive request.', ['transport_names' => $this->worker->getMetadata()->getTransportNames()]);
 
-            $this->worker->keepalive();
+            $this->worker->keepalive($this->getApplication()->getAlarmInterval());
 
             return false;
         }

diff --git a/src/Symfony/Component/Messenger/Command/FailedMessagesRetryCommand.php b/src/Symfony/Component/Messenger/Command/FailedMessagesRetryCommand.php
@@ -156,7 +156,7 @@ public function handleSignal(int $signal, int|false $previousExitCode = 0): int|
         if (\SIGALRM === $signal) {
             $this->logger?->info('Sending keepalive request.', ['transport_names' => $this->worker->getMetadata()->getTransportNames()]);
 
-            $this->worker->keepalive();
+            $this->worker->keepalive($this->getApplication()->getAlarmInterval());
 
             return false;
         }

diff --git a/src/Symfony/Component/Messenger/Tests/WorkerTest.php b/src/Symfony/Component/Messenger/Tests/WorkerTest.php
@@ -641,7 +641,7 @@ public function testKeepalive()
 
         try {
             $oldAsync = pcntl_async_signals(true);
-            pcntl_signal(\SIGALRM, fn () => $worker->keepalive());
+            pcntl_signal(\SIGALRM, fn () => $worker->keepalive(2));
             pcntl_alarm(2);
 
             $worker->run();
@@ -654,7 +654,7 @@ public function testKeepalive()
         $this->assertSame($expectedEnvelopes, $receiver->keepaliveEnvelopes);
 
         $receiver->keepaliveEnvelopes = [];
-        $worker->keepalive();
+        $worker->keepalive(2);
 
         $this->assertCount(0, $receiver->keepaliveEnvelopes);
     }
@@ -672,7 +672,7 @@ class DummyKeepaliveReceiver extends DummyReceiver implements KeepaliveReceiverI
 {
     public array $keepaliveEnvelopes = [];
 
-    public function keepalive(Envelope $envelope): void
+    public function keepalive(Envelope $envelope, ?int $seconds = null): void
     {
         $this->keepaliveEnvelopes[] = $envelope;
     }

diff --git a/src/Symfony/Component/Messenger/Transport/Receiver/KeepaliveReceiverInterface.php b/src/Symfony/Component/Messenger/Transport/Receiver/KeepaliveReceiverInterface.php
@@ -19,7 +19,9 @@ interface KeepaliveReceiverInterface extends ReceiverInterface
     /**
      * Informs the transport that the message is still being processed to avoid a timeout on the transport's side.
      *
+     * @param int|null $seconds The minimum duration the message should be kept alive
+     *
      * @throws TransportException If there is an issue communicating with the transport
      */
-    public function keepalive(Envelope $envelope): void;
+    public function keepalive(Envelope $envelope, ?int $seconds = null): void;
 }
diff --git a/src/Symfony/Component/Messenger/Worker.php b/src/Symfony/Component/Messenger/Worker.php
@@ -290,7 +290,7 @@ public function stop(): void
         $this->shouldStop = true;
     }
 
-    public function keepalive(): void
+    public function keepalive(?int $seconds): void
     {
         foreach ($this->keepalives as $message) {
             [$transportName, $envelope] = $this->keepalives[$message];
@@ -303,7 +303,7 @@ public function keepalive(): void
                 'transport' => $transportName,
                 'message_id' => $envelope->last(TransportMessageIdStamp::class)?->getId(),
             ]);
-            $this->receivers[$transportName]->keepalive($envelope);
+            $this->receivers[$transportName]->keepalive($envelope, $seconds);
         }
     }