feat(amqp): make amqp v2 compatible

BedrockStreaming · Feb 28, 2024 · 626b630 · 626b630
1 parent ffe058e
commit 626b630
Show file tree

Hide file tree

Showing 3 changed files with 25 additions and 7 deletions.
diff --git a/src/AmqpBundle/Amqp/Consumer.php b/src/AmqpBundle/Amqp/Consumer.php
@@ -63,7 +63,13 @@ public function ackMessage(string $deliveryTag, int $flags = AMQP_NOPARAM): bool
             $this->eventDispatcher->dispatch($ackEvent,AckEvent::NAME);
         }
 
-        return $this->call($this->queue, 'ack', [$deliveryTag, $flags]);
+        $isAcked = $this->call($this->queue, 'ack', [$deliveryTag, $flags]);
+
+        if ($isAcked === null) {
+            return true;
+        }
+
+        return $isAcked;
     }
 
     /**
@@ -83,7 +89,13 @@ public function nackMessage(string $deliveryTag, int $flags = AMQP_NOPARAM): boo
             $this->eventDispatcher->dispatch($nackEvent, NackEvent::NAME);
         }
 
-        return $this->call($this->queue, 'nack', [$deliveryTag, $flags]);
+        $isNacked = $this->call($this->queue, 'nack', [$deliveryTag, $flags]);
+
+        if ($isNacked === null) {
+            return true;
+        }
+
+        return $isNacked;
     }
 
     /**
@@ -100,7 +112,13 @@ public function purge(): bool
             $this->eventDispatcher->dispatch($purgeEvent, PurgeEvent::NAME);
         }
 
-        return $this->call($this->queue, 'purge');
+        $isPurged = $this->call($this->queue, 'purge');
+
+        if ($isPurged === null) {
+            return true;
+        }
+
+        return $isPurged;
     }
 
     /**

diff --git a/src/AmqpBundle/Amqp/Producer.php b/src/AmqpBundle/Amqp/Producer.php
@@ -63,13 +63,13 @@ public function publishMessage(string $message, int $flags = AMQP_NOPARAM, array
         }
 
         if (!$routingKeys) {
-            return $this->call($this->exchange, 'publish', [$message, null, $flags, $attributes]);
+            return $this->call($this->exchange, 'publish', [$message, null, $flags, $attributes]) === null;
         }
 
         // Publish the message for each routing keys
         $success = true;
         foreach ($routingKeys as $routingKey) {
-            $success &= $this->call($this->exchange, 'publish', [$message, $routingKey, $flags, $attributes]);
+            $success &= $this->call($this->exchange, 'publish', [$message, $routingKey, $flags, $attributes]) === null;
         }
 
         return (bool) $success;

diff --git a/src/AmqpBundle/Tests/Units/Amqp/Consumer.php b/src/AmqpBundle/Tests/Units/Amqp/Consumer.php
@@ -221,8 +221,8 @@ public function testPurge()
             ->and($queue = $this->getQueue($msgList))
             ->and($consumer = new Base($queue, []))
                 // Purge the queue
-                ->boolean($consumer->purge())
-                    ->isTrue()
+                ->integer($consumer->purge())
+                    ->isEqualTo(1)
                 ->mock($queue)
                     ->call('purge')
                         ->once()