0.0.1 version

Author: Anton Dorozhkin

.gitignore

@@ -0,0 +1,4 @@
+/vendor/
+/composer.lock
+/.php_cs.cache
+/tests/_support/_generated/

.php_cs

@@ -0,0 +1,15 @@
+<?php
+
+$finder = PhpCsFixer\Finder::create()
+    ->exclude('vendor')
+    ->in(__DIR__);
+
+return PhpCsFixer\Config::create()
+    ->setRules([
+        '@Symfony' => true,
+        'concat_space' => ['spacing' => 'one'],
+        'phpdoc_align' => false,
+        'phpdoc_to_comment' => false,
+        'header_comment' => false,
+    ])
+    ->setFinder($finder);

Makefile

@@ -0,0 +1,10 @@
+WORKING_DIR=$(CURDIR)
+
+php-cs-check:
+	$(WORKING_DIR)/vendor/bin/php-cs-fixer fix --dry-run --format=junit --diff
+
+php-cs-fix:
+	$(WORKING_DIR)/vendor/bin/php-cs-fixer fix
+
+test-unit:
+	./vendor/bin/codecept run unit

README.md

@@ -1 +1,73 @@
-# codeception-kafka
+# Codeception Kafka Extension
+
+## THIS MODULE IS NOT PRODUCTION READY
+
+This extension supports working with Apache Kafka.
+
+## Installation
+
+1. Install library
+    ```bash
+    composer require lamoda/codeception-kafka
+    ```
+   
+2. Create message serializer for your data transfer object
+
+```
+namespace Tests\KafkaModule;
+
+use App\EventBus\DtoInterface;
+use Lamoda\Codeception\Extension\MessageSerializer\MessageSerializerInterface;
+
+class AcmeMessageSerializer implements MessageSerializerInterface
+{
+    public function serialize($dto): string
+    {
+        if (!$dto instanceif DtoInterface) {
+            throw new \RuntimeException('This value must be an ' . DtoInterface::class);
+        }
+
+        $message = json_encode($dto->toArray());
+
+        if (!is_string($message)) {
+            throw new \RuntimeException(json_last_error(), json_last_error_msg());
+        }
+
+        return $message;
+    }
+}
+```
+
+The default message serializer is Lamoda\Codeception\Extension\MessageSerializer\ArrayMessageSerializer.
+
+2. Include to suite and configure
+    ```yaml
+    modules:
+        enabled:
+            - \Lamoda\Codeception\Extension\KafkaModule
+                 serializer: 'Tests\KafkaModule\AcmeMessageSerializer'
+                 config:
+                     metadata.broker.list: '192.168.99.100:9092'
+                     group.id: 'group_for_tests'
+                 topic_config:
+                     offset.store.sync.interval.ms: '0'
+                     auto.commit.interval.ms: '500'
+                     auto.offset.reset: 'smallest'   
+    ```
+
+## Development
+
+### PHP Coding Standards Fixer
+
+```bash
+make php-cs-check
+make php-cs-fix
+```
+
+### Tests
+
+Unit
+
+```bash
+make test-unit
+```

composer.json

@@ -0,0 +1,26 @@
+{
+  "name": "lamoda/codeception-kafka",
+  "description": "Kafka helper for codeception tests",
+  "type": "library",
+  "license": "MIT",
+  "minimum-stability": "stable",
+  "authors": [
+    {
+      "name": "Lamoda developers",
+      "homepage": "https://tech.lamoda.ru/"
+    }
+  ],
+  "require": {
+    "php": ">=7.1",
+    "codeception/codeception": "~2.5"
+  },
+  "autoload": {
+    "psr-4": {
+      "Lamoda\\Codeception\\Extension\\": "src/Extension/"
+    }
+  },
+  "require-dev": {
+    "friendsofphp/php-cs-fixer": "^2.13",
+    "kwn/php-rdkafka-stubs": "^1.1.0"
+  }
+}

src/Extension/KafkaModule.php

@@ -0,0 +1,202 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Lamoda\Codeception\Extension;
+
+use Codeception\Module;
+use Exception;
+use Lamoda\Codeception\Extension\MessageSerializer\MessageSerializerInterface;
+use RdKafka\Conf;
+use RdKafka\Consumer;
+use RdKafka\Message;
+use RdKafka\Producer;
+use RdKafka\Queue;
+use RdKafka\TopicConf;
+
+class KafkaModule extends Module
+{
+    protected const DEFAULT_PARTITION = 0;
+
+    /**
+     * @var MessageSerializerInterface
+     */
+    protected $messageSerializer;
+
+    /**
+     * @var Conf
+     */
+    protected $conf;
+
+    /**
+     * @var TopicConf
+     */
+    protected $topicConf;
+
+    /**
+     * @var Consumer
+     */
+    protected $consumer;
+
+    /**
+     * @var Queue
+     */
+    protected $queue;
+
+    /**
+     * @param array $settings
+     */
+    public function _beforeSuite($settings = []): void
+    {
+        parent::_beforeSuite();
+
+        if (isset($this->config['serializer']) && class_exists($this->config['serializer'])) {
+            $this->messageSerializer = new $this->config['serializer']();
+        } else {
+            $this->messageSerializer = new ArrayMessageSerializer();
+        }
+
+        $this->conf = new Conf();
+
+        if (isset($this->config['config']) && is_array($this->config['config'])) {
+            foreach ($this->config['config'] as $key => $value) {
+                $this->conf->set($key, $value);
+            }
+        }
+
+        $this->topicConf = new TopicConf();
+
+        if (isset($this->config['topic_config']) && is_array($this->config['topic_config'])) {
+            foreach ($this->config['topic_config'] as $key => $value) {
+                $this->topicConf->set($key, $value);
+            }
+        }
+
+        $this->consumer = new Consumer($this->conf);
+        $this->queue = $this->consumer->newQueue();
+    }
+
+    public function putMessageInTopic(string $topicName, string $message, ?int $partition = null): void
+    {
+        $producer = new Producer($this->conf);
+
+        $topic = $producer->newTopic($topicName, $this->topicConf);
+
+        $topic->produce($partition ?? static::DEFAULT_PARTITION, 0, $message);
+    }
+
+    public function putMessageListInTopic(string $topicName, array $messages, ?int $partition = null): void
+    {
+        foreach ($messages as $message) {
+            $this->putMessageInTopic($topicName, $message, $partition);
+        }
+    }
+
+    /**
+     * @throws Exception
+     */
+    public function seeMessageInTopic(string $topicName, string $message, ?int $partition = null): void
+    {
+        $topMessage = $this->readOneMessageByCurrentOffset($topicName, $partition ?? static::DEFAULT_PARTITION);
+
+        $this->assertNotNull($topMessage);
+        $this->assertEquals($message, $topMessage->payload);
+    }
+
+    /**
+     * @throws Exception
+     */
+    public function readAllMessagesFromTopic(string $topicName, ?int $partition = null, ?string $groupId = null): void
+    {
+        $topMessage = true;
+
+        while (null !== $topMessage) {
+            $topMessage = $this->readOneMessageByCurrentOffset(
+                $topicName,
+                $partition ?? static::DEFAULT_PARTITION,
+                $groupId
+            );
+        }
+    }
+
+    /**
+     * @param mixed $dto
+     */
+    public function putDtoInTopic(string $topicName, $dto, ?int $partition = null): void
+    {
+        $message = $this->messageSerializer->serialize($dto);
+        $this->putMessageInTopic($topicName, $message, $partition);
+    }
+
+    /**
+     * @param mixed $dto
+     *
+     * @throws Exception
+     */
+    public function seeDtoInTopic(string $topicName, $dto, ?int $partition = null): void
+    {
+        $message = $this->messageSerializer->serialize($dto);
+        $this->seeMessageInTopic($topicName, $message, $partition);
+    }
+
+    /**
+     * @throws Exception
+     */
+    public function assertTopicNotContainsUnreadMessages(string $topicName, ?int $partition = null): void
+    {
+        $this->assertNull($this->readOneMessageByCurrentOffset($topicName, $partition ?? static::DEFAULT_PARTITION));
+    }
+
+    /**
+     * @throws Exception
+     */
+    private function readOneMessageByCurrentOffset(string $topicName, int $partition, ?string $groupId = null): ?Message
+    {
+        if (null === $groupId) {
+            $consumer = $this->consumer;
+            $queue = $this->queue;
+        } else {
+            $conf = new Conf();
+            foreach ($this->config['config'] as $key => $value) {
+                $conf->set($key, $value);
+            }
+            $conf->set('group.id', $groupId);
+            $consumer = new Consumer($conf);
+            $queue = $consumer->newQueue();
+        }
+
+        $topic = $consumer->newTopic($topicName, $this->topicConf);
+        $topic->consumeQueueStart($partition, RD_KAFKA_OFFSET_STORED, $queue);
+
+        $message = $queue->consume(2000);
+
+        $topic->consumeStop($partition);
+
+        return $this->decideUponMessage($message);
+    }
+
+    /**
+     * @throws Exception
+     */
+    private function decideUponMessage(?Message $message = null): ?Message
+    {
+        if (!($message instanceof Message)) {
+            return null;
+        }
+
+        switch ($message->err) {
+            case RD_KAFKA_RESP_ERR_NO_ERROR:
+                return $message;
+                break;
+            case RD_KAFKA_RESP_ERR__PARTITION_EOF:
+                return null;
+                break;
+            case RD_KAFKA_RESP_ERR__TIMED_OUT:
+                throw new Exception('Timed out');
+                break;
+            default:
+                throw new Exception($message->errstr(), $message->err);
+                break;
+        }
+    }
+}

src/Extension/MessageSerializer/ArrayMessageSerializer.php

@@ -0,0 +1,23 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Lamoda\Codeception\Extension\MessageSerializer;
+
+class ArrayMessageSerializer implements MessageSerializerInterface
+{
+    public function serialize($dto): string
+    {
+        if (!is_array($dto)) {
+            throw new \RuntimeException('This value must be an array');
+        }
+
+        $message = json_encode($dto);
+
+        if (!is_string($message)) {
+            throw new \RuntimeException(json_last_error(), json_last_error_msg());
+        }
+
+        return $message;
+    }
+}

src/Extension/MessageSerializer/MessageSerializerInterface.php

@@ -0,0 +1,10 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Lamoda\Codeception\Extension\MessageSerializer;
+
+interface MessageSerializerInterface
+{
+    public function serialize($dto);
+}