Fixing issue where transport didn't ack if message has additional stamps

- If an envelope was wrapped with sendable stamps from the moment
  they were received till acked/received, then the system would
  be unable to remove the item from the processing queue
- Added docker support for php to make testing easier

Signed-off-by: RJ Garcia <[email protected]>

Author: RJ Garcia

Dockerfile

@@ -0,0 +1,8 @@
+FROM php:7.2-cli
+
+RUN apt-get update && apt-get install -y git zip
+
+COPY --from=mlocati/php-extension-installer /usr/bin/install-php-extensions /usr/bin/
+RUN install-php-extensions redis pcntl
+
+COPY --from=composer:1.9.1 /usr/bin/composer /usr/bin/composer

docker-compose.yml

@@ -1,6 +1,12 @@
 version: '3'
 
 services:
+  php:
+    build: .
+    command: "tail -f /dev/null"
+    working_dir: /var/www/html
+    volumes:
+      - ./:/var/www/html
   redis:
     image: redis
     environment: { TERM: xterm }

phpunit.xml.dist

@@ -9,7 +9,7 @@
     <php>
         <ini name="error_reporting" value="-1" />
         <env name="SHELL_VERBOSITY" value="-1" />
-        <env name="REDIS_DSN" value="redis://localhost:6379?queue=messenger"/>
+        <env name="REDIS_DSN" value="redis://redis:6379?queue=messenger"/>
     </php>
     <testsuites>
         <testsuite name="feature">

src/Transport/RedisTransport.php

@@ -6,6 +6,7 @@
 use Redis;
 use Symfony\Component\Messenger\Envelope;
 use Symfony\Component\Messenger\Stamp\DelayStamp;
+use Symfony\Component\Messenger\Stamp\TransportMessageIdStamp;
 use Symfony\Component\Messenger\Transport\Receiver\MessageCountAwareInterface;
 use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;
 use Symfony\Component\Messenger\Transport\TransportInterface;
@@ -65,10 +66,11 @@ function(Redis $conn) use ($query, $parsedUrl, $options) {
         );
     }
 
+    /** @return Envelope[] */
     public function get(): iterable {
         $this->connect();
         $this->redis->clearLastError();
-        $message = $this->redis->eval($this->popLuaScript(), [
+        $encodedMessage = $this->redis->eval($this->popLuaScript(), [
             $this->getUniqueSetName(),
             $this->getDelayedSetName(),
             $this->queue,
@@ -80,24 +82,26 @@ public function get(): iterable {
             throw new TransportException('Failed to retrieve message from queue. Redis Error: ' . $this->redis->getLastError());
         }
 
-        if (!$message) {
+        if (!$encodedMessage) {
             return [];
         }
 
-        $res = json_decode($message, true);
+        $res = json_decode($encodedMessage, true);
         $message = isset($res[0], $res[1]) ? ['body' => $res[0], 'headers' => $res[1]] : $res;
         $envelope = $this->serializer->decode($message);
-        return [$envelope];
+        return [$envelope->with(new TransportMessageIdStamp($encodedMessage))];
     }
 
     public function ack(Envelope $env): void {
         $this->connect();
-        $this->clearMessageFromProcessingQueue($this->redis, $this->encodeEnvelope($env));
+        $transportIdStamp = $env->last(TransportMessageIdStamp::class);
+        $this->clearMessageFromProcessingQueue($this->redis, $transportIdStamp ? $transportIdStamp->getId() : $this->encodeEnvelope($env));
     }
 
     public function reject(Envelope $env): void {
         $this->connect();
-        $this->clearMessageFromProcessingQueue($this->redis, $this->encodeEnvelope($env));
+        $transportIdStamp = $env->last(TransportMessageIdStamp::class);
+        $this->clearMessageFromProcessingQueue($this->redis, $transportIdStamp ? $transportIdStamp->getId() : $this->encodeEnvelope($env));
     }
 
     public function send(Envelope $env): Envelope {

tests/Feature/Fixtures/redis-config.yaml

@@ -3,9 +3,9 @@ framework:
     enabled: true
   messenger:
     transports:
-      krak_redis: 'redis://localhost:6379?queue=messenger'
+      krak_redis: 'redis://redis:6379?queue=messenger'
       sf_redis:
-        dsn: 'redis://localhost:6379'
+        dsn: 'redis://redis:6379'
         options: { use_krak_redis: false }
     routing:
       'Krak\SymfonyMessengerRedis\Tests\Feature\Fixtures\KrakRedisMessage': krak_redis

tests/Feature/TransportTest.php

@@ -6,7 +6,9 @@
 use Krak\SymfonyMessengerRedis\Transport\RedisTransport;
 use PHPUnit\Framework\TestCase;
 use Symfony\Component\Messenger\Envelope;
+use Symfony\Component\Messenger\Stamp\BusNameStamp;
 use Symfony\Component\Messenger\Stamp\DelayStamp;
+use Symfony\Component\Messenger\Stamp\ReceivedStamp;
 use Symfony\Component\Messenger\Transport\Serialization\PhpSerializer;
 use Symfony\Component\Messenger\Transport\Serialization\Serializer;
 
@@ -22,7 +24,7 @@ protected function setUp() {
         parent::setUp();
         $this->transport = RedisTransport::fromDsn(Serializer::create(), getenv('REDIS_DSN'));
         $this->redis = new \Redis();
-        $this->redis->connect('127.0.0.1');
+        $this->redis->connect('redis');
         $this->redis->flushAll();
     }
 
@@ -40,6 +42,15 @@ public function can_ack_a_message() {
         $this->then_the_queues_are_empty();
     }
 
+    /** @test */
+    public function can_ack_a_message_with_new_stamps() {
+        $this->given_there_is_a_wrapped_message();
+        $this->when_the_message_is_sent_received_and_acked(function(Envelope $env) {
+            return $env->with(new BusNameStamp('bus'));
+        });
+        $this->then_the_queues_are_empty();
+    }
+
     /** @test */
     public function can_reject_a_message() {
         $this->given_there_is_a_wrapped_message();
@@ -116,9 +127,10 @@ private function when_the_message_is_sent_on_the_transport(int $numberOfTimes =
         }
     }
 
-    private function when_the_message_is_sent_received_and_acked() {
+    private function when_the_message_is_sent_received_and_acked(callable $stampEnv = null) {
         $this->transport->send($this->envelope);
         [$env] = $this->transport->get();
+        $env = $stampEnv ? $stampEnv($env) : $env;
         $this->transport->ack($env);
     }
 
@@ -140,8 +152,8 @@ private function then_the_queue_contains_the_message() {
         $this->assertEquals(
             <<<'CONTENT'
 {"body":"{\"id\":null}","headers":{"type":"Krak\\SymfonyMessengerRedis\\Tests\\Feature\\Fixtures\\KrakRedisMessage","Content-Type":"application\/json"},"uniqueId":null}
-CONTENT,
-            $encodedMessage
+CONTENT
+            , $encodedMessage
         );
     }