fix for transactions

Author: dgafka

packages/Amqp/src/AmqpOutboundChannelAdapter.php

@@ -6,6 +6,7 @@
 
 use AMQPChannelException;
 use AMQPConnectionException;
+use Ecotone\Amqp\Transaction\AmqpTransactionInterceptor;
 use Ecotone\Enqueue\CachedConnectionFactory;
 use Ecotone\Messaging\Channel\PollableChannel\Serialization\OutboundMessageConverter;
 use Ecotone\Messaging\Conversion\ConversionService;
@@ -39,7 +40,8 @@ public function __construct(
         private bool $defaultPersistentDelivery,
         private bool $autoDeclare,
         private OutboundMessageConverter $outboundMessageConverter,
-        private ConversionService $conversionService
+        private ConversionService $conversionService,
+        private AmqpTransactionInterceptor $amqpTransactionInterceptor,
     ) {
     }
 
@@ -79,8 +81,10 @@ public function handle(Message $message): void
 
         /** @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();
+        if (!$this->amqpTransactionInterceptor->isRunningInTransaction()) {
+            /** 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())
@@ -89,7 +93,8 @@ public function handle(Message $message): void
 //            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();
+        if (!$this->amqpTransactionInterceptor->isRunningInTransaction()) {
+            $context->getExtChannel()->waitForConfirm();
+        }
     }
 }

packages/Amqp/src/AmqpOutboundChannelAdapterBuilder.php

@@ -4,6 +4,7 @@
 
 namespace Ecotone\Amqp;
 
+use Ecotone\Amqp\Transaction\AmqpTransactionInterceptor;
 use Ecotone\Enqueue\CachedConnectionFactory;
 use Ecotone\Enqueue\EnqueueOutboundChannelAdapterBuilder;
 use Ecotone\Messaging\Channel\PollableChannel\Serialization\OutboundMessageConverter;
@@ -127,6 +128,7 @@ public function compile(MessagingContainerBuilder $builder): Definition
             $this->autoDeclare,
             $outboundMessageConverter,
             new Reference(ConversionService::REFERENCE_NAME),
+            Reference::to(AmqpTransactionInterceptor::class),
         ]);
     }
 }

packages/Amqp/src/AmqpReconnectableConnectionFactory.php

@@ -37,7 +37,10 @@ public function createContext(): Context
             $this->reconnect();
         }
 
-        return $this->connectionFactory->createContext();
+        $context = $this->connectionFactory->createContext();
+        $context->getExtChannel()->setConfirmCallback(fn() => false, fn() => throw new \RuntimeException("Message was failed to be persisted in RabbitMQ instance. Check RabbitMQ server logs."));
+
+        return $context;
     }
 
     public function getConnectionInstanceId(): string

packages/Amqp/src/Transaction/AmqpTransactionInterceptor.php

@@ -90,7 +90,9 @@ public function transactional(
                 $this->logger->info(
                     'AMQP transaction was roll backed',
                     $message,
-                    $exception
+                    [
+                        'exception' => $exception
+                    ]
                 );
 
                 throw $exception;
@@ -104,4 +106,9 @@ public function transactional(
         $this->isRunningTransaction = false;
         return $result;
     }
+
+    public function isRunningInTransaction(): bool
+    {
+        return $this->isRunningTransaction;
+    }
 }

packages/Amqp/tests/Configuration/AmqpModuleTest.php

@@ -48,39 +48,6 @@ public function test_registering_amqp_backed_message_channel()
         );
     }
 
-    public function test_registering_amqp_backed_message_channel_with_application_media_type()
-    {
-        $amqpChannelBuilder = AmqpBackedMessageChannelBuilder::create('amqpChannel');
-        $messagingSystem    = MessagingSystemConfiguration::prepareWithDefaults(
-            InMemoryModuleMessaging::createWith(
-                [AmqpModule::create(InMemoryAnnotationFinder::createEmpty(), InterfaceToCallRegistry::createEmpty())],
-                [
-                    ServiceConfiguration::createWithDefaults()
-                        ->withDefaultSerializationMediaType(MediaType::APPLICATION_JSON),
-                    $amqpChannelBuilder,
-                ]
-            )
-        )
-            ->registerMessageChannel($amqpChannelBuilder)
-            ->registerConverter(new ArrayToJsonConverterBuilder())
-            ->buildMessagingSystemFromConfiguration(
-                InMemoryReferenceSearchService::createWith(
-                    [
-                        AmqpConnectionFactory::class => $this->getCachedConnectionFactory(),
-                    ]
-                )
-            );
-
-        /** @var PollableChannel $channel */
-        $channel = $messagingSystem->getMessageChannelByName('amqpChannel');
-        $channel->send(MessageBuilder::withPayload([1, 2, 3])->setContentType(MediaType::createApplicationXPHPArray())->build());
-
-        $this->assertEquals(
-            '[1,2,3]',
-            $channel->receive()->getPayload()
-        );
-    }
-
     public function test_registering_amqp_configuration()
     {
         $amqpExchange = AmqpExchange::createDirectExchange('exchange');

packages/Amqp/tests/Integration/GeneralAmqpTest.php

@@ -47,7 +47,9 @@ public function test_messages_are_delivered_after_lost_heartbeat(): void
     {
         $ecotone = $this->bootstrapEcotone(
             namespaces: ['Test\Ecotone\Amqp\Fixture\Order'],
-            services: [new OrderService(), new OrderErrorHandler(), 'logger' => new EchoLogger()],
+            services: [new OrderService(), new OrderErrorHandler(),
+//                'logger' => new EchoLogger()
+            ],
             amqpConfig: ['heartbeat' => 2]
         );