Providers.php

<?php

declare(strict_types=1);

namespace ThroughputDemo;

use Ecotone\Amqp\Publisher\AmqpMessagePublisherConfiguration;
use Ecotone\Dbal\Configuration\DbalMessagePublisherConfiguration;
use Ecotone\Kafka\Configuration\KafkaBrokerConfiguration;
use Ecotone\Kafka\Configuration\KafkaPublisherConfiguration;
use Ecotone\Lite\EcotoneLite;
use Ecotone\Messaging\Config\ModulePackageList;
use Ecotone\Messaging\Config\ServiceConfiguration;
use Ecotone\Messaging\MessagePublisher;
use Ecotone\Redis\Configuration\RedisMessagePublisherConfiguration;
use Ecotone\Sqs\Configuration\SqsMessagePublisherConfiguration;
use Enqueue\AmqpExt\AmqpConnectionFactory as AmqpExtConnectionFactory;
use Enqueue\AmqpLib\AmqpConnectionFactory as AmqpLibConnectionFactory;
use Enqueue\Dbal\DbalConnectionFactory;
use Interop\Amqp\AmqpConnectionFactory as AmqpConnectionFactoryInterface;
use Enqueue\Redis\RedisConnectionFactory;
use Enqueue\Sqs\SqsConnectionFactory;

/**
 * Bootstraps one Ecotone MessagePublisher per provider, with High Throughput
 * Publishing either off (the synchronous baseline) or on.
 *
 * High Throughput Publishing is a single switch that turns on every mechanism
 * the provider can actually deliver:
 *
 *   - batching                   — all five providers coalesce a BatchMessage
 *                                  into one round trip
 *   - non blocking confirmation  — Kafka, RabbitMQ and SQS hand back a Future
 *                                  instead of blocking on the broker's reply
 *
 * Redis and Postgres confirm the write in the reply to the very command that
 * performs it, so there is no confirmation to defer — their configuration
 * takes no parameters and only batching is on offer.
 *
 * High Throughput Publishing is an Ecotone Enterprise feature, so every
 * bootstrap is handed the licence key from ECOTONE_LICENCE_KEY.
 */
final class Providers
{
    public const KAFKA = 'kafka';
    public const SQS = 'sqs';
    public const REDIS = 'redis';
    public const DBAL = 'postgres';

    /**
     * RabbitMQ is measured twice, once per Enqueue transport, because Ecotone's
     * batch path differs between them: only an enqueue/amqp-lib context gets the
     * single socket write (batch_basic_publish() + publish_batch()). On
     * enqueue/amqp-ext each message is written separately and only the publisher
     * confirms are coalesced.
     */
    public const AMQP_EXT = 'rabbitmq-ext';
    public const AMQP_LIB = 'rabbitmq-lib';

    /**
     * Providers whose adapter can hand back a Future instead of blocking on the
     * broker's delivery confirmation. Redis and Postgres cannot: the reply that
     * confirms the write is the reply to the write itself.
     */
    public const SUPPORTS_NON_BLOCKING_CONFIRMATION = [self::KAFKA, self::AMQP_EXT, self::AMQP_LIB, self::SQS];

    public function __construct(private string $licenceKey)
    {
    }

    public function publisher(string $provider, bool $highThroughputPublishing): MessagePublisher
    {
        $queueName = self::uniqueDestination();

        [$modulePackage, $services, $publisherConfiguration] = match ($provider) {
            self::KAFKA => [
                ModulePackageList::KAFKA_PACKAGE,
                [KafkaBrokerConfiguration::class => KafkaBrokerConfiguration::createWithDefaults([self::kafkaDsn()])],
                KafkaPublisherConfiguration::createWithDefaults(topicName: $queueName),
            ],
            self::AMQP_EXT, self::AMQP_LIB => [
                ModulePackageList::AMQP_PACKAGE,
                // AmqpExtConnectionFactory::class is only the conventional reference
                // name Ecotone resolves the connection under — either transport can
                // be registered against it.
                [AmqpExtConnectionFactory::class => self::amqpConnectionFactory($provider)],
                AmqpMessagePublisherConfiguration::create()
                    ->withAutoDeclareQueueOnSend(true)
                    ->withDefaultRoutingKey($queueName),
            ],
            self::SQS => [
                ModulePackageList::SQS_PACKAGE,
                [SqsConnectionFactory::class => new SqsConnectionFactory(self::sqsDsn())],
                SqsMessagePublisherConfiguration::create(queueName: $queueName),
            ],
            self::REDIS => [
                ModulePackageList::REDIS_PACKAGE,
                [RedisConnectionFactory::class => new RedisConnectionFactory(self::redisDsn())],
                RedisMessagePublisherConfiguration::create(queueName: $queueName),
            ],
            self::DBAL => [
                ModulePackageList::DBAL_PACKAGE,
                [DbalConnectionFactory::class => new DbalConnectionFactory(self::databaseDsn())],
                DbalMessagePublisherConfiguration::create(MessagePublisher::class, $queueName),
            ],
        };

        if ($highThroughputPublishing) {
            // Same call everywhere. Kafka, RabbitMQ and SQS also accept
            // batchPublishing:, nonBlockingConfirmation: and
            // confirmationTimeoutInMilliseconds: to opt out of either mechanism.
            $publisherConfiguration = $publisherConfiguration->withHighThroughputPublishing();
        }

        if ($provider === self::AMQP_EXT || $provider === self::AMQP_LIB) {
            self::declareAmqpQueue($provider, $queueName);
        }

        return $this->bootstrap($modulePackage, $services, [$publisherConfiguration])
            ->getMessagePublisher();
    }

    private function bootstrap(
        string $modulePackage,
        array $services,
        array $extensionObjects,
    ): \Ecotone\Messaging\Config\ConfiguredMessagingSystem {
        return EcotoneLite::bootstrap(
            containerOrAvailableServices: $services,
            configuration: ServiceConfiguration::createWithDefaults()
                ->withSkippedModulePackageNames(ModulePackageList::allPackagesExcept([$modulePackage]))
                ->withNamespaces(['ThroughputDemo'])
                ->withExtensionObjects($extensionObjects),
            pathToRootCatalog: __DIR__ . '/..',
            allowGatewaysToBeRegisteredInContainer: true,
            licenceKey: $this->licenceKey !== '' ? $this->licenceKey : null,
        );
    }

    private static function declareAmqpQueue(string $provider, string $queueName): void
    {
        $context = self::amqpConnectionFactory($provider)->createContext();
        $context->declareQueue($context->createQueue($queueName));
    }

    private static function amqpConnectionFactory(string $provider): AmqpConnectionFactoryInterface
    {
        $config = ['dsn' => getenv('RABBIT_HOST') ?: 'amqp://guest:guest@localhost:5672/%2f'];

        return $provider === self::AMQP_LIB
            ? new AmqpLibConnectionFactory($config)
            : new AmqpExtConnectionFactory($config);
    }

    private static function kafkaDsn(): string
    {
        return getenv('KAFKA_DSN') ?: 'localhost:9092';
    }

    private static function sqsDsn(): string
    {
        return getenv('SQS_DSN') ?: 'sqs:?key=key&secret=secret&region=us-east-1&endpoint=http://localhost:4566&version=latest';
    }

    private static function redisDsn(): string
    {
        return getenv('REDIS_DSN') ?: 'redis+phpredis://localhost:6379';
    }

    private static function databaseDsn(): string
    {
        return getenv('DATABASE_DSN') ?: 'pgsql://ecotone:secret@localhost:5432/ecotone';
    }

    private static function uniqueDestination(): string
    {
        return uniqid('throughput_demo_');
    }
}