README

Reusable Yii2 Kafka worker and publisher package for PHP 7.4+.

Install

composer require uzmak/yii2-kafka-7-4

For local path repository:

{
  "repositories": [
    {
      "type": "path",
      "url": "../Composer-package/yii2-kafka-7.4"
    }
  ]
}

Requirements

• PHP 7.4+
• ext-rdkafka
• Yii2

App Config

'components' => [
    'kafka' => [
        'class' => Uzmak\Yii2Kafka74\KafkaConfigProvider::class,
        'brokers' => '127.0.0.1:9092',
        'username' => 'username',
        'password' => 'secret',
        'securityProtocol' => 'SASL_PLAINTEXT',
        'saslMechanisms' => 'PLAIN',
        'autoOffsetReset' => 'earliest',
        'autoCommit' => true,
        'consumeTimeoutMs' => 1000,
        'retryMaxAttempts' => 3,
        'retryBackoffMs' => 500,
        'producerFlushRetries' => 3,
        'producerFlushTimeoutMs' => 10000,
        'debug' => '',
        'logLevel' => 4,
    ],
],
'controllerMap' => [
    'worker' => [
        'class' => Uzmak\Yii2Kafka74\controllers\WorkerController::class,
        'handlerPath' => '@common/kafka/handlers',
        'handlerNamespace' => 'common\\kafka\\handlers',
    ],
    'kafka-publish' => [
        'class' => Uzmak\Yii2Kafka74\controllers\KafkaPublishController::class,
    ],
],

If broker does not require auth:

'kafka' => [
    'class' => Uzmak\Yii2Kafka74\KafkaConfigProvider::class,
    'brokers' => 'kafka:9092',
    'username' => null,
    'password' => null,
],

Handler Example

<?php

namespace common\kafka\handlers;

use Uzmak\Yii2Kafka74\Contracts\KafkaHandlerInterface;

class OrderCreateHandler implements KafkaHandlerInterface
{
    public function getTopic(): string
    {
        return 'order-create';
    }

    public function getGroup(): string
    {
        return 'order-service';
    }

    public function handle($payload): void
    {
        echo json_encode($payload, JSON_UNESCAPED_UNICODE | JSON_PRETTY_PRINT) . PHP_EOL;
    }
}

Commands

Minimal logs:

php yii worker/start

Full logs:

php yii worker/start --vvv=1

Daemon mode:

php yii worker/start --daemon=1

Publish one message:

php yii kafka-publish/send order-create '{"order_id":999}'

Publish batch:

php yii kafka-publish/batch order-create '[{"id":1},{"id":2}]'

Use Publisher In Service/Class

You can use KafkaPublisher directly inside any Yii2 service or plain PHP class.

Service example:

<?php

namespace common\services;

use Yii;
use Uzmak\Yii2Kafka74\KafkaPublisher;
use Uzmak\Yii2Kafka74\Producer;

class OrderEventService
{
    private $publisher;

    public function __construct()
    {
        $this->publisher = new KafkaPublisher(
            new Producer(Yii::$app->kafka->createOptions(false))
        );
    }

    public function publishCreated(array $order): void
    {
        $this->publisher->publishSend('order-create', json_encode([
            'order_id' => $order['id'],
            'status' => $order['status'],
        ], JSON_UNESCAPED_UNICODE));
    }

    public function publishCreatedBatch(array $orders): void
    {
        $messages = [];

        foreach ($orders as $order) {
            $messages[] = [
                'order_id' => $order['id'],
                'status' => $order['status'],
            ];
        }

        $this->publisher->publishBatch(
            'order-create',
            json_encode($messages, JSON_UNESCAPED_UNICODE)
        );
    }
}

publishSend() bitta JSON object yuboradi:

$service->publishCreated([
    'id' => 101,
    'status' => 'new',
]);

publishBatch() JSON objectlardan iborat array yuboradi:

$service->publishCreatedBatch([
    ['id' => 101, 'status' => 'new'],
    ['id' => 102, 'status' => 'paid'],
]);

Plain class example:

<?php

namespace common\components;

use Yii;
use Uzmak\Yii2Kafka74\Producer;

class AuditKafkaSender
{
    private $producer;

    public function __construct()
    {
        $this->producer = new Producer(Yii::$app->kafka->createOptions(false));
    }

    public function send(array $payload): void
    {
        $this->producer->send('audit-log', $payload);
    }

    public function sendBatch(array $payloads): void
    {
        foreach ($payloads as $payload) {
            $this->producer->send('audit-log', $payload);
        }
    }
}

Plain class ichida:

$sender->send([
    'action' => 'login',
    'user_id' => 12,
]);

$sender->sendBatch([
    ['action' => 'login', 'user_id' => 12],
    ['action' => 'logout', 'user_id' => 12],
]);

Migrations

php yii migrate --migrationPath=@vendor/uzmak/yii2-kafka-7-4/src/migrations

Creates kafka_failed_event table for failed messages.