add publish confirm

Author: lexxksb

composer.json

@@ -17,7 +17,9 @@
     }
   ],
   "require": {
-    "php": ">=7.2"
+    "php": ">=7.2",
+    "ext-libxml": "*",
+    "ext-amqp": "*"
   },
   "autoload-dev": {
     "classmap": [

src/Protocols/Amqp.php

@@ -8,6 +8,9 @@
 use AMQPEnvelope;
 use AMQPExchange;
 use AMQPQueue;
+use AMQPQueueException;
+use AMQPException;
+use AMQPBasicProperties;
 use Tochka\MQAdapter\Exceptions\MqAdapterException;
 
 /**
@@ -16,14 +19,25 @@
  */
 class Amqp
 {
+    public const CONNECT_TYPE_PUBLISH = 'publish';
+    public const CONNECT_TYPE_CONSUME = 'consume';
+    public const PUBLISH_CONFIRM_WAIT = 0.4;
+    
     protected $hosts;
     protected $login;
     protected $password;
     protected $settings;
+    /**
+     * @var AMQPConnection $connection
+     */
     protected $connection;
+    /**
+     * @var AMQPChannel $channel
+     */
     protected $channel;
     protected $queues;
     protected $currentQueue;
+    protected $connectType = self::CONNECT_TYPE_CONSUME;
 
     /**
      * AmqpAdapter constructor.
@@ -41,6 +55,11 @@ public function __construct($hosts, $login, $password, $settings)
         $this->settings = $settings;
     }
 
+    public function setConnectTypePublish(): void
+    {
+        $this->connectType = self::CONNECT_TYPE_PUBLISH;
+    }
+    
     /**
      * При уничтожении объекта - отключаемся
      */
@@ -86,14 +105,25 @@ public function disconnect(): void
      */
     public function send(string $destination, string $message, array $settings = []): bool
     {
-        $this->checkConnection();
+        $this->setConnectTypePublish();
 
-        $exchange = new AMQPExchange($this->channel);
+        $this->checkConnection();
 
-        $result = $exchange->publish($message, $destination, AMQP_NOPARAM, $settings);
+        $this->setPublisherConfirms();
 
+        $exchange = new AMQPExchange($this->channel);
+        $exchange->setName($destination);
+        $result = $exchange->publish($message, '', AMQP_NOPARAM, $settings);
         if (!$result) {
-            throw new \AMQPException('Failed to publish content.');
+            throw new AMQPException('Failed to publish content.');
+        }
+        
+        try {
+            $this->channel->waitForConfirm(self::PUBLISH_CONFIRM_WAIT);
+        } catch (AMQPException $e) {
+            if ((int)$e->getCode() !== 0) {
+                throw $e;
+            }
         }
 
         return $result;
@@ -102,7 +132,8 @@ public function send(string $destination, string $message, array $settings = [])
     /**
      * Вычитывает и возвращает новые сообщения (если они есть)
      *
-     * @return bool|AMQPEnvelope
+     * @return AMQPEnvelope|array
+     * @throws AMQPQueueException
      */
     public function getNextMessage()
     {
@@ -114,8 +145,15 @@ public function getNextMessage()
 
         $this->currentQueue = $queue;
 
-        /** @var AMQPEnvelope $message */
-        $message = $queue->get(AMQP_NOPARAM);
+        try {
+            /** @var AMQPEnvelope $message */
+            $message = $queue->get(AMQP_NOPARAM);
+        } catch (AMQPQueueException $e) {
+            if ((int)$e->getCode() === 404) {
+                sleep(3);
+            }
+            throw $e;
+        }
 
         if (is_bool($message)) {
             return $message;
@@ -163,7 +201,7 @@ public function subscribeAll(): void
         }
     }
 
-    public function unsubscribe($queueName)
+    public function unsubscribe($queueName): void
     {
         if (empty($this->queues[$queueName])) {
             return;
@@ -200,7 +238,7 @@ public function __sleep()
      *
      * @return array
      */
-    protected function adaptMessage(AMQPEnvelope $message)
+    protected function adaptMessage(AMQPEnvelope $message): array
     {
         $headers = $message->getHeaders();
         $headers['destination'] = $this->currentQueue->getName();
@@ -234,6 +272,10 @@ protected function hasErrors(): bool
 
     /**
      * Выполняет переподключение
+     *
+     * @return bool
+     * @throws MqAdapterException
+     * @throws \AMQPConnectionException
      */
     protected function reconnect(): bool
     {
@@ -249,6 +291,9 @@ protected function reconnect(): bool
 
             $this->connection->reconnect();
             $this->channel = new AMQPChannel($this->connection);
+            if ($this->connectType === self::CONNECT_TYPE_PUBLISH) {
+                $this->setPublisherConfirms();
+            }
 
             if (isset($queues)) {
                 $this->clearSubscribes();
@@ -263,6 +308,29 @@ protected function reconnect(): bool
         return $this->connect();
     }
 
+    /**
+     * Включаем механизм подтверждения публикации сообщения
+     */
+    private function setPublisherConfirms(): void
+    {
+        
+        $this->channel->setPrefetchCount(1);
+        $this->channel->confirmSelect();
+        $this->channel->setConfirmCallback(function ($delivery_tag, $multiple) {
+            echo 'Message acked: ' . $delivery_tag . '/' . ($multiple ? 'multiple' : 'noMultiple'), PHP_EOL;
+            
+            return true;
+        }, function ($delivery_tag, $multiple, $requeue) {
+            throw new MqAdapterException('Message nacked: ' . $delivery_tag . '/' . ($multiple ? 'multiple' : 'noMultiple') . '/' . $requeue);
+        });
+        $this->channel->setReturnCallback(
+            function ($reply_code, $reply_text, $exchange, $routing_key, AMQPBasicProperties $properties, $body) {
+                echo 'Message returned: ', $reply_code, '-', $reply_text, ', message body:', $body, PHP_EOL;
+            }
+        );
+        
+    }
+    
     /**
      * @return bool
      * @throws MqAdapterException
@@ -306,6 +374,10 @@ protected function connect(): bool
 
                 $this->channel = new AMQPChannel($link);
 
+                if ($this->connectType === self::CONNECT_TYPE_PUBLISH) {
+                    $this->setPublisherConfirms();
+                }
+                
                 return true;
             }
         }