refactor: split High Throughput Publishing into batching and non blocking confirmation (Enterprise)

Monorepo/Benchmark/AsyncPublishingBenchmark.php → Monorepo/Benchmark/HighThroughputPublishingBenchmark.php

@@ -42,7 +42,7 @@
  * and command reply respectively), so the asynchronous scenarios are not supported for them and are not benchmarked.
  */
 #[Warmup(0), Revs(1), Iterations(10)]
-class AsyncPublishingBenchmark
+class HighThroughputPublishingBenchmark
 {
     private const AMOUNT_OF_PUBLISHED_MESSAGES = 1000;
 
@@ -58,13 +58,13 @@ class AsyncPublishingBenchmark
 
     public function setUpAmqpSynchronousPublishing(): void
     {
-        $this->publisher = $this->bootstrapAmqpPublisher(asyncPublishing: false);
+        $this->publisher = $this->bootstrapAmqpPublisher(highThroughputPublishing: false);
         $this->warmUpPublisher();
     }
 
-    public function setUpAmqpAsyncPublishing(): void
+    public function setUpAmqpHighThroughputPublishing(): void
     {
-        $this->publisher = $this->bootstrapAmqpPublisher(asyncPublishing: true);
+        $this->publisher = $this->bootstrapAmqpPublisher(highThroughputPublishing: true);
         $this->warmUpPublisher();
     }
 
@@ -80,13 +80,13 @@ public function setUpAmqpBatchChannel(): void
 
     public function setUpKafkaSynchronousPublishing(): void
     {
-        $this->publisher = $this->bootstrapKafkaPublisher(asyncPublishing: false);
+        $this->publisher = $this->bootstrapKafkaPublisher(highThroughputPublishing: false);
         $this->warmUpPublisher();
     }
 
-    public function setUpKafkaAsyncPublishing(): void
+    public function setUpKafkaHighThroughputPublishing(): void
     {
-        $this->publisher = $this->bootstrapKafkaPublisher(asyncPublishing: true);
+        $this->publisher = $this->bootstrapKafkaPublisher(highThroughputPublishing: true);
         $this->warmUpPublisher();
     }
 
@@ -135,13 +135,13 @@ public function setUpRedisBatchChannel(): void
 
     public function setUpSqsSynchronousPublishing(): void
     {
-        $this->publisher = $this->bootstrapSqsPublisher(asyncPublishing: false);
+        $this->publisher = $this->bootstrapSqsPublisher(highThroughputPublishing: false);
         $this->warmUpPublisher();
     }
 
-    public function setUpSqsAsyncPublishing(): void
+    public function setUpSqsHighThroughputPublishing(): void
     {
-        $this->publisher = $this->bootstrapSqsPublisher(asyncPublishing: true);
+        $this->publisher = $this->bootstrapSqsPublisher(highThroughputPublishing: true);
         $this->warmUpPublisher();
     }
 
@@ -161,7 +161,7 @@ public function bench_amqp_single_message_synchronous(): void
         $this->publishSynchronouslyOneByOne();
     }
 
-    #[BeforeMethods('setUpAmqpAsyncPublishing')]
+    #[BeforeMethods('setUpAmqpHighThroughputPublishing')]
     public function bench_amqp_single_message_asynchronous(): void
     {
         $this->publishAsynchronouslyOneByOne();
@@ -173,7 +173,7 @@ public function bench_amqp_batch_message_synchronous(): void
         $this->publishBatchSynchronously();
     }
 
-    #[BeforeMethods('setUpAmqpAsyncPublishing')]
+    #[BeforeMethods('setUpAmqpHighThroughputPublishing')]
     public function bench_amqp_batch_message_asynchronous(): void
     {
         $this->publishBatchAsynchronously();
@@ -185,7 +185,7 @@ public function bench_amqp_multiple_batches_synchronous(): void
         $this->publishMultipleBatchesSynchronously();
     }
 
-    #[BeforeMethods('setUpAmqpAsyncPublishing')]
+    #[BeforeMethods('setUpAmqpHighThroughputPublishing')]
     public function bench_amqp_multiple_batches_asynchronous(): void
     {
         $this->publishMultipleBatchesAsynchronously();
@@ -197,7 +197,7 @@ public function bench_kafka_single_message_synchronous(): void
         $this->publishSynchronouslyOneByOne();
     }
 
-    #[BeforeMethods('setUpKafkaAsyncPublishing')]
+    #[BeforeMethods('setUpKafkaHighThroughputPublishing')]
     public function bench_kafka_single_message_asynchronous(): void
     {
         $this->publishAsynchronouslyOneByOne();
@@ -209,7 +209,7 @@ public function bench_kafka_batch_message_synchronous(): void
         $this->publishBatchSynchronously();
     }
 
-    #[BeforeMethods('setUpKafkaAsyncPublishing')]
+    #[BeforeMethods('setUpKafkaHighThroughputPublishing')]
     public function bench_kafka_batch_message_asynchronous(): void
     {
         $this->publishBatchAsynchronously();
@@ -221,7 +221,7 @@ public function bench_kafka_multiple_batches_synchronous(): void
         $this->publishMultipleBatchesSynchronously();
     }
 
-    #[BeforeMethods('setUpKafkaAsyncPublishing')]
+    #[BeforeMethods('setUpKafkaHighThroughputPublishing')]
     public function bench_kafka_multiple_batches_asynchronous(): void
     {
         $this->publishMultipleBatchesAsynchronously();
@@ -257,7 +257,7 @@ public function bench_sqs_single_message_synchronous(): void
         $this->publishSynchronouslyOneByOne();
     }
 
-    #[BeforeMethods('setUpSqsAsyncPublishing')]
+    #[BeforeMethods('setUpSqsHighThroughputPublishing')]
     public function bench_sqs_single_message_asynchronous(): void
     {
         $this->publishAsynchronouslyOneByOne();
@@ -269,7 +269,7 @@ public function bench_sqs_batch_message_synchronous(): void
         $this->publishBatchSynchronously();
     }
 
-    #[BeforeMethods('setUpSqsAsyncPublishing')]
+    #[BeforeMethods('setUpSqsHighThroughputPublishing')]
     public function bench_sqs_batch_message_asynchronous(): void
     {
         $this->publishBatchAsynchronously();
@@ -281,7 +281,7 @@ public function bench_sqs_multiple_batches_synchronous(): void
         $this->publishMultipleBatchesSynchronously();
     }
 
-    #[BeforeMethods('setUpSqsAsyncPublishing')]
+    #[BeforeMethods('setUpSqsHighThroughputPublishing')]
     public function bench_sqs_multiple_batches_asynchronous(): void
     {
         $this->publishMultipleBatchesAsynchronously();
@@ -298,7 +298,7 @@ private function publishAsynchronouslyOneByOne(): void
     {
         $futures = [];
         for ($messageNumber = 0; $messageNumber < self::AMOUNT_OF_PUBLISHED_MESSAGES; $messageNumber++) {
-            $futures[] = $this->publisher->asyncPublish(self::MESSAGE_PAYLOAD, MediaType::TEXT_PLAIN);
+            $futures[] = $this->publisher->publishDeferred(self::MESSAGE_PAYLOAD, MediaType::TEXT_PLAIN);
         }
         foreach ($futures as $future) {
             $future->resolve();
@@ -314,7 +314,7 @@ private function publishBatchSynchronously(): void
 
     private function publishBatchAsynchronously(): void
     {
-        $this->publisher->asyncPublish($this->buildBatch(self::AMOUNT_OF_PUBLISHED_MESSAGES), MediaType::TEXT_PLAIN)->resolve();
+        $this->publisher->publishDeferred($this->buildBatch(self::AMOUNT_OF_PUBLISHED_MESSAGES), MediaType::TEXT_PLAIN)->resolve();
     }
 
     private function publishMultipleBatchesSynchronously(): void
@@ -330,7 +330,7 @@ private function publishMultipleBatchesAsynchronously(): void
     {
         $futures = [];
         for ($batchNumber = 0; $batchNumber < self::AMOUNT_OF_BATCHES; $batchNumber++) {
-            $futures[] = $this->publisher->asyncPublish($this->buildBatch(self::MESSAGES_PER_BATCH), MediaType::TEXT_PLAIN);
+            $futures[] = $this->publisher->publishDeferred($this->buildBatch(self::MESSAGES_PER_BATCH), MediaType::TEXT_PLAIN);
         }
         foreach ($futures as $future) {
             $future->resolve();
@@ -375,7 +375,7 @@ private function bootstrapBatchChannel(string $modulePackage, object $channelBui
         return $messaging->getMessageChannel($channelBuilder->getMessageChannelName());
     }
 
-    private function bootstrapAmqpPublisher(bool $asyncPublishing): MessagePublisher
+    private function bootstrapAmqpPublisher(bool $highThroughputPublishing): MessagePublisher
     {
         $queueName = uniqid('benchmark_orders_');
         $connectionFactory = new AmqpConnectionFactory(['dsn' => getenv('RABBIT_HOST') ?: 'amqp://guest:guest@localhost:5672/%2f']);
@@ -385,8 +385,8 @@ private function bootstrapAmqpPublisher(bool $asyncPublishing): MessagePublisher
         $publisherConfiguration = AmqpMessagePublisherConfiguration::create()
             ->withAutoDeclareQueueOnSend(true)
             ->withDefaultRoutingKey($queueName);
-        if ($asyncPublishing) {
-            $publisherConfiguration = $publisherConfiguration->withAsyncPublishing();
+        if ($highThroughputPublishing) {
+            $publisherConfiguration = $publisherConfiguration->withHighThroughputPublishing();
         }
 
         $messaging = EcotoneLite::bootstrapFlowTesting(
@@ -435,11 +435,11 @@ private function bootstrapRedisPublisher(): MessagePublisher
         return $messaging->getGateway(MessagePublisher::class);
     }
 
-    private function bootstrapSqsPublisher(bool $asyncPublishing): MessagePublisher
+    private function bootstrapSqsPublisher(bool $highThroughputPublishing): MessagePublisher
     {
         $publisherConfiguration = SqsMessagePublisherConfiguration::create(queueName: uniqid('benchmark_orders_'));
-        if ($asyncPublishing) {
-            $publisherConfiguration = $publisherConfiguration->withAsyncPublishing();
+        if ($highThroughputPublishing) {
+            $publisherConfiguration = $publisherConfiguration->withHighThroughputPublishing();
         }
 
         $messaging = EcotoneLite::bootstrapFlowTesting(
@@ -456,11 +456,11 @@ private function bootstrapSqsPublisher(bool $asyncPublishing): MessagePublisher
         return $messaging->getGateway(MessagePublisher::class);
     }
 
-    private function bootstrapKafkaPublisher(bool $asyncPublishing): MessagePublisher
+    private function bootstrapKafkaPublisher(bool $highThroughputPublishing): MessagePublisher
     {
         $publisherConfiguration = KafkaPublisherConfiguration::createWithDefaults(topicName: uniqid('benchmark_orders_'));
-        if ($asyncPublishing) {
-            $publisherConfiguration = $publisherConfiguration->withAsyncPublishing();
+        if ($highThroughputPublishing) {
+            $publisherConfiguration = $publisherConfiguration->withHighThroughputPublishing();
         }
 
         $messaging = EcotoneLite::bootstrapFlowTesting(

packages/Amqp/src/AmqpBackedMessageChannelBuilder.php

@@ -30,7 +30,7 @@ private function __construct(
                 ->withDefaultRoutingKey($queueName)
                 ->withAutoDeclareOnSend(true)
                 ->withDefaultPersistentMode(true)
-                ->withAsyncPublishingChannelName($channelName)
+                ->withPublishingChannelName($channelName)
         );
     }
 
@@ -72,9 +72,14 @@ public function withPublisherConfirms(bool $enabled): self
         return $this;
     }
 
-    public function withHighThroughputPublishing(bool $enabled = true, ?int $timeoutInMilliseconds = null): self
+    /**
+     * @param bool $batchPublishing coalesces published Messages into a single publisher confirms round trip
+     * @param bool $nonBlockingConfirmation publishes without waiting for publisher confirms, which are awaited before the surrounding Command Bus or asynchronous endpoint finishes
+     * @param int|null $confirmationTimeoutInMilliseconds how long to await publisher confirms before treating the delivery as failed
+     */
+    public function withHighThroughputPublishing(bool $batchPublishing = true, bool $nonBlockingConfirmation = true, ?int $confirmationTimeoutInMilliseconds = null): self
     {
-        $this->getAmqpOutboundChannelAdapter()->withAsyncPublishing($enabled, $timeoutInMilliseconds);
+        $this->getAmqpOutboundChannelAdapter()->withHighThroughputPublishing($batchPublishing, $nonBlockingConfirmation, $confirmationTimeoutInMilliseconds);
 
         return $this;
     }
@@ -93,7 +98,7 @@ public function getMessageChannelName(): string
 
     protected function supportsBatchMessages(): bool
     {
-        return $this->getAmqpOutboundChannelAdapter()->isAsyncPublishingEnabled();
+        return $this->getAmqpOutboundChannelAdapter()->isBatchPublishingEnabled();
     }
 
     public function getQueueName()

packages/Amqp/src/AmqpOutboundChannelAdapter.php

@@ -7,9 +7,9 @@
 use Ecotone\Amqp\Transaction\AmqpTransactionInterceptor;
 use Ecotone\Enqueue\CachedConnectionFactory;
 use Ecotone\Messaging\BatchMessage;
-use Ecotone\Messaging\Channel\AsyncPublishing\AsyncPublishingRegistry;
-use Ecotone\Messaging\Channel\AsyncPublishing\FailedDelivery;
-use Ecotone\Messaging\Channel\AsyncPublishing\PublishingFailedException;
+use Ecotone\Messaging\Channel\DeliveryConfirmation\FailedDelivery;
+use Ecotone\Messaging\Channel\DeliveryConfirmation\PendingDeliveryRegistry;
+use Ecotone\Messaging\Channel\DeliveryConfirmation\PublishingFailedException;
 use Ecotone\Messaging\Channel\PollableChannel\Serialization\OutboundMessageConverter;
 use Ecotone\Messaging\Config\ConfigurationException;
 use Ecotone\Messaging\Conversion\ConversionService;
@@ -54,10 +54,11 @@ public function __construct(
         private OutboundMessageConverter   $outboundMessageConverter,
         private ConversionService          $conversionService,
         private AmqpTransactionInterceptor $amqpTransactionInterceptor,
-        private AsyncPublishingRegistry    $asyncPublishingRegistry,
+        private PendingDeliveryRegistry    $pendingDeliveryRegistry,
         private ?DelayStrategy             $delayStrategy = null,
-        private bool                       $asyncPublishing = false,
-        private int                        $asyncPublishingTimeout = AmqpOutboundChannelAdapterBuilder::DEFAULT_ASYNC_PUBLISHING_TIMEOUT,
+        private bool                       $batchPublishing = false,
+        private bool                       $nonBlockingConfirmation = false,
+        private int                        $confirmationTimeout = AmqpOutboundChannelAdapterBuilder::DEFAULT_CONFIRMATION_TIMEOUT,
         private string                     $channelName = '',
     ) {
     }
@@ -68,8 +69,8 @@ public function __construct(
     public function handle(Message $message): void
     {
         $payload = $message->getPayload();
-        if ($payload instanceof BatchMessage && ! $this->asyncPublishing) {
-            throw ConfigurationException::create(sprintf('Sending BatchMessage over `%s` requires async publishing to be enabled. Enable it with withAsyncPublishing(), available as part of Ecotone Enterprise.', $this->channelName !== '' ? $this->channelName : $this->exchangeName));
+        if ($payload instanceof BatchMessage && ! $this->batchPublishing) {
+            throw ConfigurationException::create(sprintf('Sending BatchMessage over `%s` requires batch publishing to be enabled. Enable it with withHighThroughputPublishing(), available as part of Ecotone Enterprise.', $this->channelName !== '' ? $this->channelName : $this->exchangeName));
         }
 
         $messagesToPublish = $payload instanceof BatchMessage
@@ -89,7 +90,7 @@ public function handle(Message $message): void
 
         $publishRecords = $this->publishMessages($messagesToPublish, $context, $confirmations);
 
-        if ($publishRecords !== [] && $confirmations !== null && $this->canPublishAsynchronously()) {
+        if ($publishRecords !== [] && $confirmations !== null && $this->canDeferConfirmation()) {
             $this->registerPendingDelivery($publishRecords, $context, $confirmations, $prePublishConfirmationsEpoch);
 
             return;
@@ -98,9 +99,9 @@ public function handle(Message $message): void
         $this->awaitPublisherConfirmsSynchronously($publishRecords, $context, $confirmations, $prePublishConfirmationsEpoch);
     }
 
-    public function isAsyncPublishingEnabled(): bool
+    public function isNonBlockingConfirmationEnabled(): bool
     {
-        return $this->asyncPublishing;
+        return $this->nonBlockingConfirmation;
     }
 
     /**
@@ -267,24 +268,24 @@ private function prepareInteropMessage(Message $message): array
         return [$messageToSend, $exchangeName, $outboundMessage->getDeliveryDelay(), $timeToLive];
     }
 
-    private function canPublishAsynchronously(): bool
+    private function canDeferConfirmation(): bool
     {
-        return $this->asyncPublishing
+        return $this->nonBlockingConfirmation
             && $this->publisherConfirms
-            && $this->asyncPublishingRegistry->isScopeActive();
+            && $this->pendingDeliveryRegistry->isScopeActive();
     }
 
     /**
      * @param array<int, array{message: Message, deliveryTag: int, correlationId: string}> $publishRecords
      */
     private function registerPendingDelivery(array $publishRecords, InteropAmqpContext $context, AmqpPublisherConfirmations $confirmations, int $prePublishConfirmationsEpoch): void
     {
-        $this->asyncPublishingRegistry->register(
+        $this->pendingDeliveryRegistry->register(
             $this->channelName,
             new AmqpPendingDelivery(
                 $context,
                 $publishRecords,
-                $this->asyncPublishingTimeout,
+                $this->confirmationTimeout,
                 $this->channelName,
                 $confirmations,
                 $prePublishConfirmationsEpoch,
@@ -302,7 +303,7 @@ private function awaitPublisherConfirmsSynchronously(array $publishRecords, Inte
         }
 
         if ($publishRecords === [] || $confirmations === null) {
-            $timeoutInSeconds = $this->asyncPublishingTimeout / 1000;
+            $timeoutInSeconds = $this->confirmationTimeout / 1000;
             if ($context instanceof AmqpLibContext) {
                 $context->getLibChannel()->wait_for_pending_acks_returns($timeoutInSeconds);
             } elseif ($context instanceof AmqpExtContext) {
@@ -315,7 +316,7 @@ private function awaitPublisherConfirmsSynchronously(array $publishRecords, Inte
         $deliveryResult = (new AmqpPendingDelivery(
             $context,
             $publishRecords,
-            $this->asyncPublishingTimeout,
+            $this->confirmationTimeout,
             $this->channelName,
             $confirmations,
             $prePublishConfirmationsEpoch,
@@ -325,7 +326,7 @@ private function awaitPublisherConfirmsSynchronously(array $publishRecords, Inte
             return;
         }
 
-        if ($this->asyncPublishing) {
+        if ($this->nonBlockingConfirmation) {
             throw PublishingFailedException::withFailedDeliveries($deliveryResult->getFailedDeliveries());
         }