ack

Author: dgafka

packages/Amqp/src/AmqpOutboundChannelAdapter.php

@@ -11,6 +11,7 @@
 use Ecotone\Messaging\Conversion\ConversionService;
 use Ecotone\Messaging\Message;
 use Ecotone\Messaging\MessageHandler;
+use Enqueue\AmqpExt\AmqpContext;
 use Interop\Amqp\AmqpMessage;
 use Interop\Amqp\Impl\AmqpTopic;
 
@@ -76,11 +77,19 @@ public function handle(Message $message): void
         $messageToSend
             ->setDeliveryMode($this->defaultPersistentDelivery ? AmqpMessage::DELIVERY_MODE_PERSISTENT : AmqpMessage::DELIVERY_MODE_NON_PERSISTENT);
 
+        /** @var AmqpContext $context */
+        $context = $this->connectionFactory->createContext();
+        /** Ensures no messages are lost along the way when heartbeat is lost and ensures messages was peristed on the Broker side. Without this message can be simply "swallowed" without throwing exception */
+        $context->getExtChannel()->confirmSelect();
+
         $this->connectionFactory->getProducer()
             ->setTimeToLive($outboundMessage->getTimeToLive())
             ->setDelayStrategy(new HeadersExchangeDelayStrategy())
             ->setDeliveryDelay($outboundMessage->getDeliveryDelay())
 //            this allow for having queue per delay instead of queue per delay + exchangeName
             ->send(new AmqpTopic($exchangeName), $messageToSend);
+
+        $context->getExtChannel()->setConfirmCallback(fn() => false, fn() => throw new \RuntimeException("Message was not sent to exchange {$exchangeName} with routing key {$routingKey}"));
+        $context->getExtChannel()->waitForConfirm();
     }
 }

packages/Ecotone/src/Messaging/Channel/PollableChannel/SendRetries/SendRetryChannelInterceptor.php

@@ -41,7 +41,7 @@ public function afterSendCompletion(Message $message, MessageChannel $messageCha
         if ($exception !== null) {
             $attempt = 1;
             while ($this->retryTemplate->canBeCalledNextTime($attempt)) {
-                $this->logger->info("Message was not sent to {$this->relatedChannel} due to exception. Trying to self-heal by doing retry attempt: {$attempt}/{$this->retryTemplate->getMaxAttempts()}", [
+                $this->logger->info("Message was not sent to {$this->relatedChannel} due to exception. Trying to self-heal by doing retry attempt: {$attempt}/{$this->retryTemplate->getMaxAttempts()}. Exception message: `{$exception->getMessage()}`", [
                     'exception' => $exception->getMessage(),
                     'relatedChannel' => $this->relatedChannel,
                 ]);
@@ -60,7 +60,7 @@ public function afterSendCompletion(Message $message, MessageChannel $messageCha
             }
         }
 
-        $this->logger->error("Message was not sent to {$this->relatedChannel} due to exception. No more retries will be done", [
+        $this->logger->error("Message was not sent to {$this->relatedChannel} due to exception. No more retries will be done. Exception message: `{$exception->getMessage()}`", [
             'exception' => $exception->getMessage(),
             'relatedChannel' => $this->relatedChannel,
         ]);