diff --git a/.circleci/config.yml b/.circleci/config.yml index d952d1c..bcee842 100644 --- a/.circleci/config.yml +++ b/.circleci/config.yml @@ -3,11 +3,14 @@ orbs: wait-for: cobli/wait-for@0.0.2 jobs: build: + parameters: + php-version: + type: string environment: CC_TEST_REPORTER_ID: ac6b0ffb95381a6e67239f27805263c12cdb85bc1b8f3bc295173a8cb12ae2b1 working_directory: ~/messenger-kafka docker: - - image: circleci/php:7.4.4 + - image: cimg/php:<< parameters.php-version >> - image: lensesio/fast-data-dev environment: SAMPLEDATA: 0 @@ -17,9 +20,9 @@ jobs: steps: - checkout - - run: sudo apt update # PHP CircleCI 2.0 Configuration File# PHP CircleCI 2.0 Configuration File sudo apt install zlib1g-dev libsqlite3-dev + - run: sudo apt update - run: sudo apt install librdkafka1 librdkafka-dev kafkacat netcat - - run: sudo docker-php-ext-install zip && sudo pecl install rdkafka && sudo docker-php-ext-enable rdkafka + - run: sudo pecl install rdkafka xdebug # Download and cache dependencies - restore_cache: @@ -55,13 +58,22 @@ jobs: - run: name: PHPUnit command: | - mkdir results && mkdir results/tests ./cc-test-reporter before-build - ./vendor/bin/simple-phpunit --log-junit results/tests/junit.xml --coverage-html results/coverage-report --coverage-clover clover.xml + XDEBUG_MODE=coverage ./vendor/bin/simple-phpunit --log-junit junit.xml --coverage-clover clover.xml ./cc-test-reporter after-build --coverage-input-type clover --exit-code $? - store_test_results: - path: ~/messenger-kafka/results/tests + path: ~/messenger-kafka/junit.xml - - store_artifacts: - path: ~/messenger-kafka/results/coverage-report \ No newline at end of file + - store_test_results: + path: ~/messenger-kafka/clover.xml + +workflows: + all-versions: + jobs: + - build: + name: "php-7.4" + php-version: "7.4" + - build: + name: "php-8.1" + php-version: "8.1" \ No newline at end of file diff --git a/.github/dependabot.yml b/.github/dependabot.yml new file mode 100644 index 0000000..a51bb0b --- /dev/null +++ b/.github/dependabot.yml @@ -0,0 +1,11 @@ +# To get started with Dependabot version updates, you'll need to specify which +# package ecosystems to update and where the package manifests are located. +# Please see the documentation for all configuration options: +# https://help.github.com/github/administering-a-repository/configuration-options-for-dependency-updates + +version: 2 +updates: + - package-ecosystem: "composer" # See documentation for possible values + directory: "/" # Location of package manifests + schedule: + interval: "daily" diff --git a/.github/workflows/php.yml b/.github/workflows/php.yml new file mode 100644 index 0000000..5b59d26 --- /dev/null +++ b/.github/workflows/php.yml @@ -0,0 +1,78 @@ +name: Tests + +on: + push: + branches: [ master ] + pull_request: + branches: [ master ] + +jobs: + build: + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + php: ['8.3', '8.4', '8.5'] + symfony: ['^6.4', '^7.3', '^7.4', '^8.0'] + exclude: + - php: '8.3' + symfony: '^8.0' + - php: '8.3' + symfony: '^7.4' + + services: + kafka: + image: lensesio/fast-data-dev + env: + ADV_HOST: 127.0.0.1 + SAMPLEDATA: 0 + RUNTESTS: 0 + FORWARDLOGS: 0 + ports: + - 9092:9092 + - 8081:8081 + - 8082:8082 + - 8083:8083 + - 2181:2181 + + steps: + - uses: actions/checkout@v2 + + - name: Validate composer.json and composer.lock + run: composer validate + + - name: Install rdkafka + run: + sudo add-apt-repository universe && sudo apt update && sudo apt install netcat-traditional librdkafka1 librdkafka-dev + + - name: Setup PHP + uses: shivammathur/setup-php@v2 + with: + php-version: "${{ matrix.php }}" + coverage: xdebug + tools: pecl, phpunit + extensions: rdkafka + ini-values: post_max_size=256M, short_open_tag=On + + - name: check if rdkafka is there + run: php -v && php --ri rdkafka + + - name: Install dependencies + run: | + composer global require --no-progress --no-scripts --no-plugins "symfony/flex:^2.4" + composer global config --no-plugins allow-plugins.symfony/flex true + composer config extra.symfony.require "${{ matrix.symfony }}" + composer update --prefer-dist --no-progress + + - name: Wait for Kafka + run: .github/workflows/wait-for-kafka.sh + + - name: Unit Tests + run: ./vendor/bin/simple-phpunit --log-junit results/tests/junit.xml + + #- name: try to produce + # run: echo "test" > testfile.json && kafkacat -b 0.0.0.0:9092 -t kafkacat_test -P testfile.json + + # Docs: https://getcomposer.org/doc/articles/scripts.md + #- name: Run test suite + # run: phpunit --log-junit results/tests/junit.xml --coverage-html results/coverage-report --coverage-clover clover.xml diff --git a/.github/workflows/wait-for-kafka.sh b/.github/workflows/wait-for-kafka.sh new file mode 100755 index 0000000..cb863d0 --- /dev/null +++ b/.github/workflows/wait-for-kafka.sh @@ -0,0 +1,11 @@ +#! /bin/bash +end=$((SECONDS+60)) + +while [ $SECONDS -lt $end ]; do + if echo dump | nc localhost 2181 | grep broker ; then + exit 0 + else + echo "Kafka didn't start in time" + fi + sleep 1 +done \ No newline at end of file diff --git a/.gitignore b/.gitignore index f27070a..a6fce38 100644 --- a/.gitignore +++ b/.gitignore @@ -9,3 +9,5 @@ composer.lock /var /coverage .DS_Store +/results +/.php-cs-fixer.cache diff --git a/.php-cs-fixer.dist.php b/.php-cs-fixer.dist.php new file mode 100644 index 0000000..13d64a2 --- /dev/null +++ b/.php-cs-fixer.dist.php @@ -0,0 +1,28 @@ +in(__DIR__) + ->exclude('var') + ->notPath('config/bundles.php') + ->notPath('config/preload.php') +; + +$config = new PhpCsFixer\Config(); +$config + ->setRiskyAllowed(true) + ->setRules([ + '@Symfony' => true, + '@Symfony:risky' => true, + '@PhpCsFixer:risky' => true, + 'array_syntax' => ['syntax' => 'short'], + 'yoda_style' => false, + 'single_import_per_statement' => false, + 'concat_space' => ['spacing' => 'one'], + 'declare_strict_types' => true, + ]) + ->setFinder($finder) + ; + +return $config; \ No newline at end of file diff --git a/.php_cs.dist b/.php_cs.dist deleted file mode 100644 index 9415beb..0000000 --- a/.php_cs.dist +++ /dev/null @@ -1,25 +0,0 @@ -in(__DIR__)->exclude(['vendor'])->append([__FILE__]); - -$config = Config::create() - ->setRiskyAllowed(true) - ->setUsingCache(true) - ->setCacheFile(__DIR__ . '/.php_cs.cache') - ->setRules([ - '@Symfony' => true, - '@PHP71Migration' => true, - 'array_indentation' => true, - 'array_syntax' => ['syntax' => 'short'], - 'concat_space' => ['spacing' => 'one'], - 'ordered_imports' => ['imports_order' => ['class', 'function', 'const']], - 'visibility_required' => ['elements' => ['property', 'method', 'const']], - ]) - ->setFinder($finder); - -return $config; diff --git a/Messenger/KafkaTransport.php b/Messenger/KafkaTransport.php deleted file mode 100644 index e73c737..0000000 --- a/Messenger/KafkaTransport.php +++ /dev/null @@ -1,188 +0,0 @@ -logger = $logger; - $this->serializer = $serializer; - $this->rdKafkaFactory = $rdKafkaFactory; - $this->kafkaConf = $kafkaConf; - $this->topicName = $topicName; - $this->flushTimeoutMs = $flushTimeoutMs; - $this->receiveTimeoutMs = $receiveTimeoutMs; - $this->commitAsync = $commitAsync; - - $this->subscribed = false; - } - - public function get(): iterable - { - $message = $this->getSubscribedConsumer()->consume($this->receiveTimeoutMs); - - switch ($message->err) { - case RD_KAFKA_RESP_ERR_NO_ERROR: - $this->logger->info(sprintf('Kafka: Message %s %s %s received ', $message->topic_name, $message->partition, $message->offset)); - - $envelope = $this->serializer->decode([ - 'body' => $message->payload, - 'headers' => property_exists($message, 'headers') ? $message->headers : null, - ]); - - return [$envelope->with(new KafkaMessageStamp($message))]; - case RD_KAFKA_RESP_ERR__PARTITION_EOF: - $this->logger->info('Kafka: Partition EOF reached. Waiting for next message ...'); - break; - case RD_KAFKA_RESP_ERR__TIMED_OUT: - $this->logger->debug('Kafka: Consumer timeout.'); - break; - case RD_KAFKA_RESP_ERR__TRANSPORT: - $this->logger->error('Kafka: Broker transport failure.'); - throw new KafkaException($message->errstr(), $message->err); - default: - $this->logger->error('Kafka: error during message consume.', [ - 'kafkaError' => $message->err - ]); - throw new KafkaException($message->errstr(), $message->err); - } - - return []; - } - - public function getSubscribedConsumer(): KafkaConsumer - { - $consumer = $this->getConsumer(); - - if (false === $this->subscribed) { - $consumer->subscribe([$this->topicName]); - $this->logger->info('Partition assignment...'); - - $this->subscribed = true; - } - - return $consumer; - } - - public function ack(Envelope $envelope): void - { - $consumer = $this->getConsumer(); - - /** @var KafkaMessageStamp $transportStamp */ - $transportStamp = $envelope->last(KafkaMessageStamp::class); - $message = $transportStamp->getMessage(); - - if ($this->commitAsync) { - $consumer->commitAsync($message); - } else { - $consumer->commit($message); - } - - $this->logger->info(sprintf('Message %s %s %s ack successful.', $message->topic_name, $message->partition, $message->offset)); - } - - public function reject(Envelope $envelope): void - { - // Do nothing. auto commit should be set to false! - } - - public function send(Envelope $envelope): Envelope - { - $producer = $this->getProducer(); - $topic = $producer->newTopic($this->topicName); - - $payload = $this->serializer->encode($envelope); - - $topic->producev(RD_KAFKA_PARTITION_UA, 0, $payload['body'], null, $payload['headers'] ?? null); - - $status = $producer->flush($this->flushTimeoutMs); - if ($status !== 0) { - $this->logger->error('Kafka: Unable to send a message to the topic.', [ - 'kafkaTopic' => $this->topicName, - 'kafkaError' => $status, - ]); - throw new KafkaException(sprintf( - 'Kafka: Unable to send a message to the topic %s, kafka error %d.', - $this->topicName, - $status - )); - } - - return $envelope; - } - - private function getConsumer(): KafkaConsumer - { - if ($this->consumer) { - return $this->consumer; - } - - $this->consumer = $this->rdKafkaFactory->createConsumer($this->kafkaConf); - - return $this->consumer; - } - - private function getProducer(): KafkaProducer - { - if ($this->producer) { - return $this->producer; - } - - $this->producer = $this->rdKafkaFactory->createProducer($this->kafkaConf); - - return $this->producer; - } -} diff --git a/Messenger/KafkaTransportFactory.php b/Messenger/KafkaTransportFactory.php deleted file mode 100644 index f70aa78..0000000 --- a/Messenger/KafkaTransportFactory.php +++ /dev/null @@ -1,129 +0,0 @@ -logger = $logger; - } - - public function supports(string $dsn, array $options): bool - { - foreach (self::DSN_PROTOCOLS as $protocol) { - if (0 === strpos($dsn, $protocol)) { - return true; - } - } - - return false; - } - - public function createTransport(string $dsn, array $options, SerializerInterface $serializer): TransportInterface - { - $conf = new KafkaConf(); - - // Set a re balance callback to log partition assignments (optional) - $conf->setRebalanceCb($this->createRebalanceCb($this->logger)); - $conf->setErrorCb(function ($kafka, $err, $reason) use ($options) { - $this->logger->error('Kafka: Error during kafka produce or consume', [ - 'kafkaTopic' => $options['topic']['name'], - 'kafkaError' => $err, - 'kafkaReason' => $reason, - ]); - throw new KafkaException(sprintf( - 'Kafka: Error during kafka produce or consume topic %s. Error code: %d - Error reason: %s', - $options['topic']['name'], - $err, - $reason - )); - }); - - $brokers = $this->stripProtocol($dsn); - $conf->set('metadata.broker.list', implode(',', $brokers)); - - foreach ($options['kafka_conf'] ?? [] as $option => $value) { - $conf->set($option, $value); - } - - foreach ($options['topic_conf'] ?? [] as $option => $value) { - $conf->set($option, $value); - } - - return new KafkaTransport( - $this->logger, - $serializer, - new RdKafkaFactory(), - $conf, - $options['topic']['name'], - $options['flushTimeout'] ?? 10000, - $options['receiveTimeout'] ?? 10000, - $options['commitAsync'] ?? false - ); - } - - private function stripProtocol(string $dsn): array - { - $brokers = []; - foreach (explode(',', $dsn) as $currentBroker) { - foreach (self::DSN_PROTOCOLS as $protocol) { - $currentBroker = str_replace($protocol, '', $currentBroker); - } - $brokers[] = $currentBroker; - } - - return $brokers; - } - - private function createRebalanceCb(LoggerInterface $logger): \Closure - { - return function (KafkaConsumer $kafka, $err, array $topicPartitions = null) use ($logger) { - switch ($err) { - case RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS: - /** @var TopicPartition $topicPartition */ - foreach ($topicPartitions as $topicPartition) { - $logger->info(sprintf('Assign: %s %s %s', $topicPartition->getTopic(), $topicPartition->getPartition(), $topicPartition->getOffset())); - } - $kafka->assign($topicPartitions); - break; - - case RD_KAFKA_RESP_ERR__REVOKE_PARTITIONS: - /** @var TopicPartition $topicPartition */ - foreach ($topicPartitions as $topicPartition) { - $logger->info(sprintf('Assign: %s %s %s', $topicPartition->getTopic(), $topicPartition->getPartition(), $topicPartition->getOffset())); - } - $kafka->assign(null); - break; - - default: - throw new \Exception($err); - } - }; - } -} diff --git a/README.md b/README.md index 70dc6bd..36c368f 100644 --- a/README.md +++ b/README.md @@ -4,8 +4,9 @@ [![Packagist](https://img.shields.io/packagist/dt/koco/messenger-kafka.svg)](https://packagist.org/packages/koco/messenger-kafka) [![Maintainability](https://api.codeclimate.com/v1/badges/7fa3d2da6a828a676f35/maintainability)](https://codeclimate.com/github/KonstantinCodes/messenger-kafka/maintainability) [![CircleCI](https://circleci.com/gh/KonstantinCodes/messenger-kafka.svg?style=svg)](https://circleci.com/gh/KonstantinCodes/messenger-kafka) +[![Tests](https://github.com/KonstantinCodes/messenger-kafka/workflows/Tests/badge.svg)](https://github.com/KonstantinCodes/messenger-kafka/actions) -This bundle aims to provide a simple Kafka transport for Symfony Messenger. +This bundle aims to provide a simple Kafka transport for Symfony Messenger. Kafka REST Proxy support coming soon. ## Installation @@ -19,27 +20,10 @@ $ composer require koco/messenger-kafka ### Applications that don't use Symfony Flex -#### Step 1: Download the Bundle - -Open a command console, enter your project directory and execute the -following command to download the latest stable version of this bundle: - -```console -$ composer require koco/messenger-kafka -``` - -This command requires you to have Composer installed globally, as explained -in the [installation chapter](https://getcomposer.org/doc/00-intro.md) -of the Composer documentation. - -#### Step 2: Enable the Bundle - -Then, enable the bundle by adding it to the list of registered bundles +After adding the composer requirement, enable the bundle by adding it to the list of registered bundles in the `config/bundles.php` file of your project: ```php -// config/bundles.php - return [ // ... Koco\Kafka\KocoKafkaBundle::class => ['all' => true], @@ -67,6 +51,7 @@ framework: # serializer: App\Infrastructure\Messenger\MySerializer options: flushTimeout: 10000 + flushRetries: 5 topic: name: 'events' kafka_conf: @@ -96,34 +81,81 @@ framework: auto.offset.reset: 'earliest' ``` -## Custom Message Formats -If you consume messages generated by other systems, you will most likely need to implement your own Serializer. +## Serializer +You will most likely want to implement your own Serializer. Please see: [https://symfony.com/doc/current/messenger.html#serializing-messages](https://symfony.com/doc/current/messenger.html#serializing-messages) +The fields `key`, `headers`, and `body` are available in the `decode()` and `encode()` methods. + ```php 'application/json' - ]; + $record = json_decode($encodedEnvelope['body'], true); - if ($encodedEnvelope['headers']['__TypeId__'] === 'MyJavaClass') { - $headers['type'] = MyPHPClass::class; - } + return new Envelope(new ProductCreated( + $record['id'], + $record['name'], + $record['description'], + )); + } - return parent::decode([ - 'headers' => $headers, - 'body' => $encodedEnvelope['body'] - ]); + public function encode(Envelope $envelope): array + { + /** @var ProductCreated $event */ + $event = $envelope->getMessage(); + + return [ + 'key' => $event->getId(), + 'headers' => [], + 'body' => json_encode([ + 'id' => $event->getId(), + 'name' => $event->getName(), + 'description' => $event->getDescription(), + ]), + ]; } + } ``` + +## How do I work with Avro? +Same as with the basic example above, you need to build your own serializer. +Within the `decode()` and `encode()` you can make use of [flix-tech/avro-serde-php](https://github.com/flix-tech/avro-serde-php). + +## What about the Confluent Schema Registry? +To connect with Schema Registry and control various settings, you can use this bundle: + +```console +$ composer require koco/avro-regy +``` + +And configure it to match your setup: + +```yaml +avro_regy: + base_uri: '%env(SCHEMA_REGISTRY_URL)%' + file_naming_strategy: subject + options: + register_missing_schemas: true + register_missing_subjects: true + serializers: + catalogue: + schema_dir: '%kernel.project_dir%/src/Catalogue/Domain/Model/Event/Avro/' + orders: + schema_dir: '%kernel.project_dir%/src/Orders/Domain/Model/Event/Avro/' + file_naming_strategy: qualified_name + options: + register_missing_schemas: false + register_missing_subjects: false +``` + +Please see [https://github.com/KonstantinCodes/avro-regy](https://github.com/KonstantinCodes/avro-regy) for the full documentation. diff --git a/Resources/config/services.xml b/Resources/config/services.xml deleted file mode 100644 index 978549b..0000000 --- a/Resources/config/services.xml +++ /dev/null @@ -1,13 +0,0 @@ - - - - - - - - - - diff --git a/composer.json b/composer.json index b69d699..39ef666 100644 --- a/composer.json +++ b/composer.json @@ -1,33 +1,45 @@ { - "name": "datadome/messenger-kafka", + "name": "koco/messenger-kafka", "type": "symfony-bundle", "description": "Symfony Messenger Kafka Transport", "keywords": ["kafka", "symfony", "messenger", "transport", "queue", "bundle"], "license": "MIT", "require": { - "php": "^7.4 || ^8.0", + "php": "^7.4|8.*", "ext-json": "*", - "ext-rdkafka": "^5.0||^6.0", - "symfony/config": "^5.0||^6.0", - "symfony/dependency-injection": "^5.0||^6.0", - "symfony/http-kernel": "^5.0||^6.0", - "symfony/messenger": "^5.0||^6.0" + "symfony/config": "^6.4|^7.3|^8.0", + "symfony/dependency-injection": "^6.4|^7.3|^8.0", + "symfony/http-kernel": "^6.4|^7.3|^8.0", + "symfony/messenger": "^6.4|^7.3|^8.0", + "psr/http-client": "^1.0", + "psr/http-factory": "^1.0", + "psr/http-message": "^1.0", + "psr/log": "^1.0.1||^2.0 ||^3.0" + }, + "require-dev": { + "friendsofphp/php-cs-fixer": "^3.6", + "kwn/php-rdkafka-stubs": "^2.1", + "symfony/phpunit-bridge": "^6.4|^7.3|^8.0", + "symfony/framework-bundle": "^6.4|^7.3|^8.0", + "symfony/serializer": "^6.4|^7.3|^8.0", + "symfony/property-access": "^6.4|^7.0|^8.0", + "phpstan/phpstan": "^1.4.6", + "nyholm/psr7": "^1.4" + }, + "suggest": { + "ext-rdkafka": "^4.0; Needed to support Kafka connectivity", + "koco/avro-regy": "Confluent Schema Registry integration" }, "autoload": { - "psr-4": { "Koco\\Kafka\\": "" }, - "exclude-from-classmap": [ - "/Tests/" - ] + "psr-4": { "Koco\\Kafka\\": "src/" } }, "autoload-dev": { - "psr-4": { "Koco\\Kafka\\Tests\\": "Tests/" } + "psr-4": { "Koco\\Kafka\\Tests\\": "tests/" } }, - "require-dev": { - "friendsofphp/php-cs-fixer": "^2.16", - "kwn/php-rdkafka-stubs": "^2.0", - "symfony/phpunit-bridge": "^5.0||^6.0", - "symfony/framework-bundle": "^5.0||^6.0", - "symfony/serializer": "^5.0||^6.0", - "symfony/property-access": "^5.0||^6.0" + "scripts": { + "test": [ + "vendor/bin/php-cs-fixer fix -v --dry-run --stop-on-violation --using-cache=no", + "vendor/bin/simple-phpunit tests/Unit --log-junit results/tests/junit.xml" + ] } } diff --git a/phpstan.neon b/phpstan.neon new file mode 100644 index 0000000..cd9ea03 --- /dev/null +++ b/phpstan.neon @@ -0,0 +1,3 @@ +parameters: + checkMissingIterableValueType: false + checkGenericClassInNonGenericObjectType: false diff --git a/phpunit.xml b/phpunit.xml index bd3121a..3e74af5 100644 --- a/phpunit.xml +++ b/phpunit.xml @@ -1,30 +1,21 @@ - - - - - - - - - ./Tests/ - - - - - - ./ - - ./Tests - ./vendor - - - - \ No newline at end of file + + + + ./ + + + ./tests + ./vendor + + + + + + + + + ./tests/ + + + diff --git a/DependencyInjection/KocoKafkaExtension.php b/src/DependencyInjection/KocoKafkaExtension.php similarity index 64% rename from DependencyInjection/KocoKafkaExtension.php rename to src/DependencyInjection/KocoKafkaExtension.php index e8f35b7..403152c 100644 --- a/DependencyInjection/KocoKafkaExtension.php +++ b/src/DependencyInjection/KocoKafkaExtension.php @@ -1,20 +1,19 @@ load('services.xml'); + $loader = new PhpFileLoader($container, new FileLocator(__DIR__ . '/../Resources/config')); + $loader->load('services.php'); } } diff --git a/KocoKafkaBundle.php b/src/KocoKafkaBundle.php similarity index 82% rename from KocoKafkaBundle.php rename to src/KocoKafkaBundle.php index 47771d1..c764a50 100644 --- a/KocoKafkaBundle.php +++ b/src/KocoKafkaBundle.php @@ -1,5 +1,7 @@ logger = $logger; + $this->serializer = $serializer; + $this->rdKafkaFactory = $rdKafkaFactory; + $this->properties = $properties; + + $this->subscribed = false; + } + + public function get(): iterable + { + $message = $this->getSubscribedConsumer()->consume($this->properties->getReceiveTimeoutMs()); + + switch ($message->err) { + case RD_KAFKA_RESP_ERR_NO_ERROR: + $this->logger->info(\sprintf( + 'Kafka: Message %s %s %s received ', + $message->topic_name, + $message->partition, + $message->offset + )); + + $envelope = $this->serializer->decode([ + 'body' => $message->payload, + 'headers' => $message->headers ?? [], + 'key' => $message->key, + 'offset' => $message->offset, + 'timestamp' => $message->timestamp, + ]); + + return [$envelope->with(new KafkaMessageStamp($message))]; + case RD_KAFKA_RESP_ERR__PARTITION_EOF: + $this->logger->info('Kafka: Partition EOF reached. Waiting for next message ...'); + break; + case RD_KAFKA_RESP_ERR__TIMED_OUT: + $this->logger->debug('Kafka: Consumer timeout.'); + break; + case RD_KAFKA_RESP_ERR__TRANSPORT: + $this->logger->debug('Kafka: Broker transport failure.'); + break; + default: + throw new TransportException($message->errstr(), $message->err); + } + + return []; + } + + public function ack(Envelope $envelope): void + { + $consumer = $this->getConsumer(); + + /** @var KafkaMessageStamp $transportStamp */ + $transportStamp = $envelope->last(KafkaMessageStamp::class); + $message = $transportStamp->getMessage(); + + if ($this->properties->isCommitAsync()) { + $consumer->commitAsync($message); + + $this->logger->info(\sprintf( + 'Offset topic=%s partition=%s offset=%s to be committed asynchronously.', + $message->topic_name, + $message->partition, + $message->offset + )); + } else { + $consumer->commit($message); + + $this->logger->info(\sprintf( + 'Offset topic=%s partition=%s offset=%s successfully committed.', + $message->topic_name, + $message->partition, + $message->offset + )); + } + } + + public function reject(Envelope $envelope): void + { + // Do nothing. auto commit should be set to false! + } + + private function getSubscribedConsumer(): KafkaConsumer + { + $consumer = $this->getConsumer(); + + if (false === $this->subscribed) { + $this->logger->info('Partition assignment...'); + $consumer->subscribe([$this->properties->getTopicName()]); + + $this->subscribed = true; + } + + return $consumer; + } + + private function getConsumer(): KafkaConsumer + { + return $this->consumer ?? $this->consumer = $this->rdKafkaFactory->createConsumer($this->properties->getKafkaConf()); + } +} diff --git a/src/Messenger/KafkaReceiverProperties.php b/src/Messenger/KafkaReceiverProperties.php new file mode 100644 index 0000000..eb80384 --- /dev/null +++ b/src/Messenger/KafkaReceiverProperties.php @@ -0,0 +1,54 @@ +kafkaConf = $kafkaConf; + $this->topicName = $topicName; + $this->receiveTimeoutMs = $receiveTimeoutMs; + $this->commitAsync = $commitAsync; + } + + public function getKafkaConf(): KafkaConf + { + return $this->kafkaConf; + } + + public function getTopicName(): string + { + return $this->topicName; + } + + public function getReceiveTimeoutMs(): int + { + return $this->receiveTimeoutMs; + } + + public function isCommitAsync(): bool + { + return $this->commitAsync; + } +} diff --git a/src/Messenger/KafkaSender.php b/src/Messenger/KafkaSender.php new file mode 100644 index 0000000..b0c622f --- /dev/null +++ b/src/Messenger/KafkaSender.php @@ -0,0 +1,92 @@ +logger = $logger; + $this->serializer = $serializer; + $this->rdKafkaFactory = $rdKafkaFactory; + $this->properties = $properties; + } + + public function send(Envelope $envelope): Envelope + { + $producer = $this->getProducer(); + $topic = $producer->newTopic($this->properties->getTopicName()); + $payload = $this->serializer->encode($envelope); + + if (method_exists($topic, 'producev')) { + // ext-rdkafka <= 4.0.0 will fail calling `producev` on librdkafka >= 1.0.0 causing segfault + // Since we are forcing to use at least librdkafka:1.0.0, no need to check the lib version anymore + if (false !== phpversion('rdkafka') && version_compare(phpversion('rdkafka'), '4.0.0', '<')) { + trigger_error( + 'ext-rdkafka < 4.0.0 is incompatible with lib-rdkafka 1.0.0 when calling `producev`. ' . + 'Falling back to `produce` (without message headers) instead.', + E_USER_WARNING + ); + } else { + $topic->producev( + RD_KAFKA_PARTITION_UA, + 0, + $payload['body'], + $payload['key'] ?? null, + $payload['headers'] ?? null, + $payload['timestamp_ms'] ?? null + ); + + $this->producer->poll(0); + } + } else { + $topic->produce( + RD_KAFKA_PARTITION_UA, + 0, + $payload['body'], + $payload['key'] ?? null + ); + + $this->producer->poll(0); + } + + for ($flushRetries = 0; $flushRetries < $this->properties->getFlushRetries() + 1; ++$flushRetries) { + $code = $producer->flush($this->properties->getFlushTimeoutMs()); + if ($code === RD_KAFKA_RESP_ERR_NO_ERROR) { + $this->logger->info(\sprintf('Kafka message sent%s', \array_key_exists('key', $payload) ? ' with key ' . $payload['key'] : '')); + break; + } + } + + if ($code !== RD_KAFKA_RESP_ERR_NO_ERROR) { + throw new TransportException('Kafka producer response error: ' . $code, $code); + } + + return $envelope; + } + + private function getProducer(): KafkaProducer + { + return $this->producer ??= $this->rdKafkaFactory->createProducer($this->properties->getKafkaConf()); + } +} diff --git a/src/Messenger/KafkaSenderProperties.php b/src/Messenger/KafkaSenderProperties.php new file mode 100644 index 0000000..f51e2e2 --- /dev/null +++ b/src/Messenger/KafkaSenderProperties.php @@ -0,0 +1,54 @@ +kafkaConf = $kafkaConf; + $this->topicName = $topicName; + $this->flushTimeoutMs = $flushTimeoutMs; + $this->flushRetries = $flushRetries; + } + + public function getKafkaConf(): KafkaConf + { + return $this->kafkaConf; + } + + public function getTopicName(): string + { + return $this->topicName; + } + + public function getFlushTimeoutMs(): int + { + return $this->flushTimeoutMs; + } + + public function getFlushRetries(): int + { + return $this->flushRetries; + } +} diff --git a/src/Messenger/KafkaTransport.php b/src/Messenger/KafkaTransport.php new file mode 100644 index 0000000..5e22e4e --- /dev/null +++ b/src/Messenger/KafkaTransport.php @@ -0,0 +1,89 @@ +logger = $logger; + $this->serializer = $serializer; + $this->rdKafkaFactory = $rdKafkaFactory; + $this->kafkaSenderProperties = $kafkaSenderProperties; + $this->kafkaReceiverProperties = $kafkaReceiverProperties; + } + + public function get(): iterable + { + return $this->getReceiver()->get(); + } + + public function ack(Envelope $envelope): void + { + $this->getReceiver()->ack($envelope); + } + + public function reject(Envelope $envelope): void + { + $this->getReceiver()->reject($envelope); + } + + public function send(Envelope $envelope): Envelope + { + return $this->getSender()->send($envelope); + } + + private function getSender(): KafkaSender + { + return $this->sender ?? $this->sender = new KafkaSender( + $this->logger, + $this->serializer, + $this->rdKafkaFactory, + $this->kafkaSenderProperties + ); + } + + private function getReceiver(): KafkaReceiver + { + return $this->receiver ?? $this->receiver = new KafkaReceiver( + $this->logger, + $this->serializer, + $this->rdKafkaFactory, + $this->kafkaReceiverProperties + ); + } +} diff --git a/src/Messenger/KafkaTransportFactory.php b/src/Messenger/KafkaTransportFactory.php new file mode 100644 index 0000000..faae40b --- /dev/null +++ b/src/Messenger/KafkaTransportFactory.php @@ -0,0 +1,123 @@ +logger = $logger ?? new NullLogger(); + $this->kafkaFactory = $kafkaFactory; + } + + public function supports(string $dsn, array $options): bool + { + foreach (self::DSN_PROTOCOLS as $protocol) { + if (0 === \strpos($dsn, $protocol)) { + return true; + } + } + + return false; + } + + public function createTransport(string $dsn, array $options, SerializerInterface $serializer): TransportInterface + { + $conf = new KafkaConf(); + + // Set a rebalance callback to log partition assignments (optional) + $conf->setRebalanceCb($this->createRebalanceCb($this->logger)); + + $brokers = $this->stripProtocol($dsn); + $conf->set('metadata.broker.list', implode(',', $brokers)); + + foreach (array_merge($options['topic_conf'] ?? [], $options['kafka_conf'] ?? []) as $option => $value) { + $conf->set($option, $value); + } + + return new KafkaTransport( + $this->logger, + $serializer, + $this->kafkaFactory, + new KafkaSenderProperties( + $conf, + $options['topic']['name'], + $options['flushTimeout'] ?? 10000, + $options['flushRetries'] ?? 0 + ), + new KafkaReceiverProperties( + $conf, + $options['topic']['name'], + $options['receiveTimeout'] ?? 10000, + $options['commitAsync'] ?? false + ) + ); + } + + private function stripProtocol(string $dsn): array + { + $brokers = []; + foreach (\explode(',', $dsn) as $currentBroker) { + foreach (self::DSN_PROTOCOLS as $protocol) { + $currentBroker = \str_replace($protocol, '', $currentBroker); + } + $brokers[] = $currentBroker; + } + + return $brokers; + } + + private function createRebalanceCb(LoggerInterface $logger): \Closure + { + return static function (KafkaConsumer $kafka, $err, ?array $topicPartitions = null) use ($logger) { + /** @var TopicPartition[] $topicPartitions */ + $topicPartitions = $topicPartitions ?? []; + + switch ($err) { + case \RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS: + foreach ($topicPartitions as $topicPartition) { + $logger->info(\sprintf('Assign: %s %s %s', $topicPartition->getTopic(), $topicPartition->getPartition(), $topicPartition->getOffset())); + } + $kafka->assign($topicPartitions); + break; + + case \RD_KAFKA_RESP_ERR__REVOKE_PARTITIONS: + foreach ($topicPartitions as $topicPartition) { + $logger->info(\sprintf('Assign: %s %s %s', $topicPartition->getTopic(), $topicPartition->getPartition(), $topicPartition->getOffset())); + } + $kafka->assign(null); + break; + + default: + throw new \Exception($err); + } + }; + } +} diff --git a/src/Messenger/RestProxySender.php b/src/Messenger/RestProxySender.php new file mode 100644 index 0000000..31bc425 --- /dev/null +++ b/src/Messenger/RestProxySender.php @@ -0,0 +1,80 @@ +baseUri = $baseUri; + $this->topicName = $topicName; + $this->serializer = $serializer; + $this->client = $client; + $this->requestFactory = $requestFactory; + $this->uriFactory = $uriFactory; + $this->streamFactory = $streamFactory; + } + + public function send(Envelope $envelope): Envelope + { + $encoded = $this->serializer->encode($envelope); + + $request = $this->requestFactory->createRequest('POST', $this->baseUri->withPath('/topics/' . $this->topicName)); + $request = $request->withHeader('Accept', 'application/vnd.kafka.v2+json'); + $request = $request->withHeader('Content-Type', $encoded['headers']['Content-Type']); + $request = $request->withBody($this->streamFactory->createStream( + '{ "records": [ { "key": "' . $encoded['key'] . '", "value": "' . $encoded['body'] . '" } ] }' + )); + + $response = $this->client->sendRequest($request); + + if ($response->getStatusCode() === 204) { + return $envelope; + } + + if ($response->getStatusCode() === 200) { + return $envelope; // $response->getBody(); TODO: do something + } + + return $envelope; + } +} diff --git a/src/Messenger/RestProxyTransport.php b/src/Messenger/RestProxyTransport.php new file mode 100644 index 0000000..44add8c --- /dev/null +++ b/src/Messenger/RestProxyTransport.php @@ -0,0 +1,106 @@ +baseUri = $baseUri; + $this->topicName = $topicName; + $this->serializer = $serializer; + $this->logger = $logger; + $this->client = $client; + $this->requestFactory = $requestFactory; + $this->uriFactory = $uriFactory; + $this->streamFactory = $streamFactory; + } + + public function get(): iterable + { + throw new \Exception('Not implemented!'); + } + + public function ack(Envelope $envelope): void + { + throw new \Exception('Not implemented!'); + } + + public function reject(Envelope $envelope): void + { + throw new \Exception('Not implemented!'); + } + + public function send(Envelope $envelope): Envelope + { + return ($this->sender ?? $this->getSender())->send($envelope); + } + + public function getMessageCount(): int + { + throw new \Exception('Not implemented!'); + } + + private function getSender(): RestProxySender + { + return $this->sender = new RestProxySender( + $this->baseUri, + $this->topicName, + $this->serializer, + $this->client, + $this->requestFactory, + $this->uriFactory, + $this->streamFactory + ); + } +} diff --git a/src/Messenger/RestProxyTransportFactory.php b/src/Messenger/RestProxyTransportFactory.php new file mode 100644 index 0000000..cff0273 --- /dev/null +++ b/src/Messenger/RestProxyTransportFactory.php @@ -0,0 +1,127 @@ +logger = $logger; + $this->client = $client; + $this->requestFactory = $requestFactory; + $this->uriFactory = $uriFactory; + $this->streamFactory = $streamFactory; + } + + public function supports(string $dsn, array $options): bool + { + return 0 === \strpos($dsn, static::DSN_PROTOCOL_KAFKA_REST); + } + + public function createTransport(string $dsn, array $options, SerializerInterface $serializer): TransportInterface + { + $this->checkDependencies(); + + $dsn = $this->uriFactory->createUri($dsn); + $scheme = $dsn->getScheme(); + + $dsnOptions = $this->queryStringToOptionsArray($dsn->getQuery()); + $options = array_merge($dsnOptions, $options); // Override DSN options with options array + + $baseUri = $dsn->withQuery(''); + + if ($scheme === static::DSN_PROTOCOL_KAFKA_REST) { + $baseUri = $baseUri->withScheme('http'); + } elseif ($scheme === static::DSN_PROTOCOL_KAFKA_REST_SSL) { + $baseUri = $baseUri->withScheme('https'); + } else { + throw new \InvalidArgumentException('The DSN is not formatted as expected.'); + } + + return new RestProxyTransport( + $baseUri, + $options['topic'], + $serializer, + $this->client, + $this->requestFactory, + $this->uriFactory, + $this->streamFactory, + $this->logger + ); + } + + private function queryStringToOptionsArray(string $queryString): array + { + $queryParts = explode('&', $queryString) ?? []; + + $dsnOptions = []; + foreach ($queryParts as $queryPart) { + list($key, $value) = explode('=', $queryPart); + $dsnOptions[$key] = urldecode($value); + } + + return $dsnOptions; + } + + private function checkDependencies(): void + { + if ($this->client === null) { + throw $this->createMissingServiceException(ClientInterface::class, 'PSR-7 HTTP Client not found.'); + } + + if ($this->requestFactory === null) { + throw $this->createMissingServiceException(RequestFactoryInterface::class, 'PSR HTTP RequestFactory not found.'); + } + + if ($this->uriFactory === null) { + throw $this->createMissingServiceException(UriFactoryInterface::class, 'PSR HTTP UriFactory not found.'); + } + + if ($this->streamFactory === null) { + throw $this->createMissingServiceException(StreamFactoryInterface::class, 'PSR HTTP StreamFactory not found.'); + } + } + + private function createMissingServiceException(string $className, ?string $message = null) + { + return new \InvalidArgumentException(\sprintf( + '%sPlease install a library that provides "%s" and ensure the service is registered.', + $message ? $message . ' ' : '', + $className + )); + } +} diff --git a/RdKafka/RdKafkaFactory.php b/src/RdKafka/RdKafkaFactory.php similarity index 93% rename from RdKafka/RdKafkaFactory.php rename to src/RdKafka/RdKafkaFactory.php index 8426d55..b22e611 100644 --- a/RdKafka/RdKafkaFactory.php +++ b/src/RdKafka/RdKafkaFactory.php @@ -1,5 +1,7 @@ services() + ->set(RdKafkaFactory::class) + ->class(RdKafkaFactory::class) + ->set(KafkaTransportFactory::class) + ->tag('messenger.transport_factory') + ->class(KafkaTransportFactory::class) + ->args([ + service(RdKafkaFactory::class), + service('logger')->nullOnInvalid(), + ]) + ->set(RestProxyTransportFactory::class) + ->tag('messenger.transport_factory') + ->class(RestProxyTransportFactory::class) + ->args([ + service('logger')->nullOnInvalid(), + service(ClientInterface::class)->nullOnInvalid(), + service(RequestFactoryInterface::class)->nullOnInvalid(), + service(UriFactoryInterface::class)->nullOnInvalid(), + service(StreamFactoryInterface::class)->nullOnInvalid(), + ]); +}; diff --git a/src/Serializer/AbstractKafkaRestProxyBinarySerializer.php b/src/Serializer/AbstractKafkaRestProxyBinarySerializer.php new file mode 100644 index 0000000..f0d17d5 --- /dev/null +++ b/src/Serializer/AbstractKafkaRestProxyBinarySerializer.php @@ -0,0 +1,31 @@ +extractKey($envelope); + + return [ + 'key' => $key ? base64_encode($key) : null, + 'body' => base64_encode($this->extractBody($envelope)), + 'headers' => ['Content-Type' => 'application/vnd.kafka.binary.v2+json'], + ]; + } + + abstract protected function extractKey(Envelope $envelope): ?string; + + abstract protected function extractBody(Envelope $envelope): string; +} diff --git a/Tests/Functional/KafkaTransportTest.php b/tests/Functional/KafkaTransportTest.php similarity index 62% rename from Tests/Functional/KafkaTransportTest.php rename to tests/Functional/KafkaTransportTest.php index d70614f..48bacd1 100644 --- a/Tests/Functional/KafkaTransportTest.php +++ b/tests/Functional/KafkaTransportTest.php @@ -1,9 +1,11 @@ testStartTime = new \DateTimeImmutable(); - } - protected function setUp(): void { /** @var LoggerInterface $logger */ $logger = $this->createMock(LoggerInterface::class); - $this->factory = new KafkaTransportFactory($logger); + $this->factory = new KafkaTransportFactory(new RdKafkaFactory(), $logger); $this->serializerMock = $this->createMock(SerializerInterface::class); ++$this->testIteration; + + $this->testStartTime = $this->testStartTime ?? new \DateTimeImmutable(); } - public function serializerProvider() + public static function provideSerializer(): \Generator { $serializer = new Serializer(); $phpSerializer = new PhpSerializer(); - return [ - [ - $serializer, - $this->createSerializerDecodeClosure($serializer), - ], - [ - $phpSerializer, - $this->createPHPSerializerDecodeClosure($phpSerializer), - ], - ]; + yield [$serializer, self::createSerializerDecodeClosure($serializer)]; + yield [$phpSerializer, self::createPHPSerializerDecodeClosure($phpSerializer)]; } /** - * @dataProvider serializerProvider - * - * @group legacy - * @expectedDeprecation Unsilenced deprecation: Function RdKafka\Conf::setDefaultTopicConf() is deprecated + * @dataProvider provideSerializer */ - public function testSendAndReceive(SerializerInterface $serializer, Closure $decodeClosure) + public function testSendAndReceive(SerializerInterface $serializer, \Closure $decodeClosure) { $sender = $this->factory->createTransport( self::BROKER, [ - 'flushTimeout' => 1000, + 'flushTimeout' => 5000, + 'flushRetries' => 5, 'topic' => [ 'name' => $this->getTopicName(), ], @@ -99,6 +86,7 @@ public function testSendAndReceive(SerializerInterface $serializer, Closure $dec 'kafka_conf' => [ 'group.id' => 'test_group', 'enable.auto.offset.store' => 'false', + 'session.timeout.ms' => '10000', ], 'topic_conf' => [ 'auto.offset.reset' => 'earliest', @@ -107,48 +95,48 @@ public function testSendAndReceive(SerializerInterface $serializer, Closure $dec $this->serializerMock ); - $this->serializerMock->expects($this->once()) + $this->serializerMock->expects(self::once()) ->method('decode') ->willReturnCallback($decodeClosure); /** @var []Envelope $envelopes */ $envelopes = $receiver->get(); - $this->assertInstanceOf(Envelope::class, $envelopes[0]); + self::assertInstanceOf(Envelope::class, $envelopes[0]); $message = $envelopes[0]->getMessage(); - $this->assertInstanceOf(TestMessage::class, $message); + self::assertInstanceOf(TestMessage::class, $message); $receiver->ack($envelopes[0]); } - public function createSerializerDecodeClosure(SerializerInterface $serializer): Closure + public static function createSerializerDecodeClosure(SerializerInterface $serializer): \Closure { - return function (array $encodedEnvelope) use ($serializer) { - $this->assertIsArray($encodedEnvelope); + return static function (array $encodedEnvelope) use ($serializer) { + self::assertIsArray($encodedEnvelope); - $this->assertEquals('{"data":"my_test_data"}', $encodedEnvelope['body']); + self::assertSame('{"data":"my_test_data"}', $encodedEnvelope['body']); - $this->assertArrayHasKey('headers', $encodedEnvelope); + self::assertArrayHasKey('headers', $encodedEnvelope); $headers = $encodedEnvelope['headers']; - $this->assertEquals(TestMessage::class, $headers['type']); - $this->assertEquals('application/json', $headers['Content-Type']); + self::assertSame(TestMessage::class, $headers['type']); + self::assertSame('application/json', $headers['Content-Type']); return $serializer->decode($encodedEnvelope); }; } - public function createPHPSerializerDecodeClosure(SerializerInterface $serializer): Closure + public static function createPHPSerializerDecodeClosure(SerializerInterface $serializer): \Closure { - return function (array $encodedEnvelope) use ($serializer) { - $this->assertIsArray($encodedEnvelope); + return static function (array $encodedEnvelope) use ($serializer) { + self::assertIsArray($encodedEnvelope); - $this->assertEquals( + self::assertSame( 'O:36:\"Symfony\\\\Component\\\\Messenger\\\\Envelope\":2:{s:44:\"\0Symfony\\\\Component\\\\Messenger\\\\Envelope\0stamps\";a:0:{}s:45:\"\0Symfony\\\\Component\\\\Messenger\\\\Envelope\0message\";O:39:\"Koco\\\\Kafka\\\\Tests\\\\Functional\\\\TestMessage\":1:{s:4:\"data\";s:12:\"my_test_data\";}}', $encodedEnvelope['body'] ); - $this->assertArrayHasKey('headers', $encodedEnvelope); + self::assertArrayHasKey('headers', $encodedEnvelope); return $serializer->decode($encodedEnvelope); }; diff --git a/Tests/Functional/TestMessage.php b/tests/Functional/TestMessage.php similarity index 86% rename from Tests/Functional/TestMessage.php rename to tests/Functional/TestMessage.php index b9f336a..3745dbb 100644 --- a/Tests/Functional/TestMessage.php +++ b/tests/Functional/TestMessage.php @@ -1,5 +1,7 @@ send() + + self::assertTrue(true); + } +} + +class DummyClient implements ClientInterface +{ + public function sendRequest(RequestInterface $request): ResponseInterface + { + echo 'hallo'; + // TODO: Implement sendRequest() method. + } +} + +class DummySerializer implements SerializerInterface +{ + public function decode(array $encodedEnvelope): Envelope + { + // TODO: Implement decode() method. + } + + public function encode(Envelope $envelope): array + { + // TODO: Implement encode() method. + } +} diff --git a/Tests/Unit/Messenger/KafkaTransportFactoryTest.php b/tests/Unit/Messenger/KafkaTransportFactoryTest.php similarity index 67% rename from Tests/Unit/Messenger/KafkaTransportFactoryTest.php rename to tests/Unit/Messenger/KafkaTransportFactoryTest.php index 82a0d7f..ef5fd07 100644 --- a/Tests/Unit/Messenger/KafkaTransportFactoryTest.php +++ b/tests/Unit/Messenger/KafkaTransportFactoryTest.php @@ -1,8 +1,11 @@ createMock(LoggerInterface::class); - $this->factory = new KafkaTransportFactory($logger); + $this->factory = new KafkaTransportFactory(new RdKafkaFactory(), $logger); $this->serializerMock = $this->createMock(SerializerInterface::class); } public function testSupports() { - $this->assertTrue($this->factory->supports('kafka://my-local-kafka:9092', [])); - $this->assertTrue($this->factory->supports('kafka+ssl://my-staging-kafka:9093', [])); - $this->assertTrue($this->factory->supports('kafka+ssl://prod-kafka-01:9093,kafka+ssl://prod-kafka-01:9093,kafka+ssl://prod-kafka-01:9093', [])); + self::assertTrue($this->factory->supports('kafka://my-local-kafka:9092', [])); + self::assertTrue($this->factory->supports('kafka+ssl://my-staging-kafka:9093', [])); + self::assertTrue($this->factory->supports('kafka+ssl://prod-kafka-01:9093,kafka+ssl://prod-kafka-01:9093,kafka+ssl://prod-kafka-01:9093', [])); } /** * @group legacy - * @expectedDeprecation Unsilenced deprecation: Function RdKafka\Conf::setDefaultTopicConf() is deprecated */ public function testCreateTransport() { @@ -52,6 +54,6 @@ public function testCreateTransport() $this->serializerMock ); - $this->assertInstanceOf(TransportInterface::class, $transport); + self::assertInstanceOf(TransportInterface::class, $transport); } } diff --git a/Tests/Unit/Messenger/KafkaTransportTest.php b/tests/Unit/Messenger/KafkaTransportTest.php similarity index 69% rename from Tests/Unit/Messenger/KafkaTransportTest.php rename to tests/Unit/Messenger/KafkaTransportTest.php index 1a9c311..ae7f90f 100644 --- a/Tests/Unit/Messenger/KafkaTransportTest.php +++ b/tests/Unit/Messenger/KafkaTransportTest.php @@ -1,8 +1,12 @@ mockLogger, $this->mockSerializer, new RdKafkaFactory(), - new KafkaConf(), - 'test', - 10000, - 10000, - false + new KafkaSenderProperties( + new KafkaConf(), + 'test', + 10000, + 10000 + ), + new KafkaReceiverProperties( + new KafkaConf(), + 'test', + 10000, + false + ) ); - $this->assertInstanceOf(TransportInterface::class, $transport); + self::assertInstanceOf(TransportInterface::class, $transport); } public function testGet() { - $this->mockRdKafkaConsumer - ->method('subscribe') - ->willReturn(true); + $this->mockRdKafkaConsumer->method('subscribe'); $testMessage = new Message(); $testMessage->err = RD_KAFKA_RESP_ERR_NO_ERROR; @@ -91,7 +100,7 @@ public function testGet() ->method('consume') ->willReturn($testMessage); - $this->mockSerializer->expects($this->once()) + $this->mockSerializer->expects(self::once()) ->method('decode') ->with([ 'body' => '{"data":null}', @@ -99,6 +108,9 @@ public function testGet() 'type' => TestMessage::class, 'Content-Type' => 'application/json', ], + 'key' => null, + 'offset' => 0, + 'timestamp' => 1586861356, ]) ->willReturn(new Envelope(new TestMessage())); @@ -106,30 +118,37 @@ public function testGet() $this->mockLogger, $this->mockSerializer, $this->mockRdKafkaFactory, - new KafkaConf(), - 'test', - 10000, - 10000, - false + new KafkaSenderProperties( + new KafkaConf(), + 'test', + 10000, + 10000 + ), + new KafkaReceiverProperties( + new KafkaConf(), + 'test', + 10000, + false + ) ); $receivedMessages = $transport->get(); - $this->assertArrayHasKey(0, $receivedMessages); + self::assertArrayHasKey(0, $receivedMessages); /** @var Envelope $receivedMessage */ $receivedMessage = $receivedMessages[0]; - $this->assertInstanceOf(Envelope::class, $receivedMessage); - $this->assertInstanceOf(TestMessage::class, $receivedMessage->getMessage()); + self::assertInstanceOf(Envelope::class, $receivedMessage); + self::assertInstanceOf(TestMessage::class, $receivedMessage->getMessage()); $stamps = $receivedMessage->all(); - $this->assertCount(1, $stamps); - $this->assertArrayHasKey(KafkaMessageStamp::class, $stamps); + self::assertCount(1, $stamps); + self::assertArrayHasKey(KafkaMessageStamp::class, $stamps); $kafkaMessageStamps = $stamps[KafkaMessageStamp::class]; - $this->assertCount(1, $kafkaMessageStamps); + self::assertCount(1, $kafkaMessageStamps); /** @var KafkaMessageStamp $kafkaMessageStamp */ $kafkaMessageStamp = $kafkaMessageStamps[0]; - $this->assertEquals($testMessage, $kafkaMessageStamp->getMessage()); + self::assertSame($testMessage, $kafkaMessageStamp->getMessage()); } } diff --git a/Tests/Unit/Messenger/TestMessage.php b/tests/Unit/Messenger/TestMessage.php similarity index 77% rename from Tests/Unit/Messenger/TestMessage.php rename to tests/Unit/Messenger/TestMessage.php index 545d5f3..0594d79 100644 --- a/Tests/Unit/Messenger/TestMessage.php +++ b/tests/Unit/Messenger/TestMessage.php @@ -1,5 +1,7 @@