feature symfony#58552 [Console][Messenger] Add $seconds to keepalive() methods (valtzu)

This PR was merged into the 7.2 branch.

Discussion
----------

[Console][Messenger] Add `$seconds` to `keepalive()` methods

| Q             | A
| ------------- | ---
| Branch?       | 7.2
| Bug fix?      | no
| New feature?  | yes
| Deprecations? | no
| License       | MIT

Make the transport aware for how long (at minimum) the message should be kept alive.

F.e. when extending SQS visibility timeout, you need to pass the visibility timeout (seconds) as parameter. If you pass a value which is less than alarm interval, SQS would resend the message too early. By making the transport aware of when the next keepalive call will happen we'll be able to do some assertions/clamping to improve DX.

This is a prerequisite for symfony#58483.

Commits
-------

e3beb18 Add `$seconds` to `keepalive` methods

Author: fabpot

src/Symfony/Component/Console/Application.php

@@ -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) {

src/Symfony/Component/Console/Tests/ApplicationTest.php

@@ -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);
     }
 

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());
     }

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

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;
         }

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;
         }

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;
     }

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;
 }

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);
         }
     }