first release

Author: mxp100

.gitignore

@@ -0,0 +1,3 @@
+/.idea
+/vendor
+composer.lock

README.md

@@ -1,2 +1,24 @@
 # kafka-queue
 Laravel Kafka Queue connection with Horizon support
+
+## Installation
+
+### 1. Composer
+```shell
+composer require laravel-tool/kafka-queue
+```
+
+### 2. Add to config/queue.php
+```php
+'kafka' => [
+    'driver' => 'kafka',
+    'broker_list' => env('KAFKA_BROKER_LIST', 'kafka:9092'),
+    'queue' => env('KAFKA_QUEUE', 'default'),
+    'heartbeat_ms' => env('KAFKA_HEARTBEAT', 5000),
+    'group_name' => env('KAFKA_QUEUE_GROUP', 'default'),
+    'producer_timeout_ms' => 1000,
+    'consumer_timeout_ms' => 3000,
+    'horizon' => true,
+    'after_commit' => false,
+],
+```

composer.json

@@ -0,0 +1,39 @@
+{
+    "name": "laravel-tool/kafka-queue",
+    "description": "Laravel Kafka Queue driver with Horizon support",
+    "keywords": [
+        "php",
+        "laravel",
+        "kafka",
+        "queue",
+        "horizon"
+    ],
+    "type": "laravel-plugin",
+    "license": "MIT",
+    "version": "1.0.0",
+    "authors": [
+        {
+            "name": "Yuriy Potemkin",
+            "email": "ysoft2000@mail.ru"
+        }
+    ],
+    "minimum-stability": "dev",
+    "require": {
+        "php": "^8.3",
+        "ext-rdkafka": "*",
+        "ext-json": "*",
+        "laravel/framework": ">=10"
+    },
+    "autoload": {
+        "psr-4": {
+            "LaravelTool\\KafkaQueue\\": "src/"
+        }
+    },
+    "extra": {
+        "laravel": {
+            "providers": [
+                "LaravelTool\\KafkaQueue\\ServiceProvider"
+            ]
+        }
+    }
+}

src/Connector.php

@@ -0,0 +1,23 @@
+<?php
+
+namespace LaravelTool\KafkaQueue;
+
+use Illuminate\Contracts\Queue\Queue as QueueInterface;
+use Illuminate\Queue\Connectors\ConnectorInterface;
+use LaravelTool\KafkaQueue\Kafka\Consumer;
+use LaravelTool\KafkaQueue\Kafka\Producer;
+
+class Connector implements ConnectorInterface
+{
+
+    public function connect(array $config): QueueInterface
+    {
+        return new Queue(
+            new Producer($config),
+            new Consumer($config),
+            defaultQueue: $config['queue'] ?? 'default',
+            horizon: $config['horizon'] ?? false,
+        );
+    }
+
+}

src/Job.php

@@ -0,0 +1,60 @@
+<?php
+
+namespace LaravelTool\KafkaQueue;
+
+use Illuminate\Container\Container;
+use Illuminate\Queue\Jobs\Job as BaseJob;
+use Illuminate\Contracts\Queue\Job as JobInterface;
+use RdKafka\Message;
+
+class Job extends BaseJob implements JobInterface
+{
+    protected array $decoded;
+
+    public function __construct(
+        Container $container,
+        protected Queue $kafkaQueue,
+        protected string $job,
+        protected Message $message,
+    ) {
+        $this->container = $container;
+        $this->decoded = $this->payload();
+        $this->connectionName = $this->kafkaQueue->getConnectionName();
+        $this->queue = $this->message->topic_name;
+    }
+
+    public function getJobId()
+    {
+        return $this->decoded['id'] ?? null;
+    }
+
+    public function getRawBody(): string
+    {
+        return $this->job;
+    }
+
+    public function attempts()
+    {
+        return ($this->decoded['attempts'] ?? null) + 1;
+    }
+
+    public function release($delay = 0): void
+    {
+        parent::release($delay);
+
+        $this->delete();
+        $this->kafkaQueue->release($delay, $this);
+    }
+
+    public function delete(): void
+    {
+        parent::delete();
+
+        $this->kafkaQueue->deleteReserved($this->queue, $this);
+    }
+
+    public function getMessageTimestamp(): int
+    {
+        return (int) ($this->message->timestamp / 1000);
+    }
+}

src/Kafka/Consumer.php

@@ -0,0 +1,66 @@
+<?php
+
+namespace LaravelTool\KafkaQueue\Kafka;
+
+use RdKafka\Conf as KafkaConfig;
+use RdKafka\Exception as KafkaException;
+use RdKafka\KafkaConsumer;
+use RdKafka\Message;
+use RuntimeException;
+
+class Consumer
+{
+    private KafkaConsumer $consumer;
+
+    public function __construct(
+        protected array $config
+    ) {
+        $this->consumer = new KafkaConsumer($this->generateConfig($config));
+    }
+
+    public function consume(string $topic): ?Message
+    {
+        try {
+            $this->checkSubscription($topic);
+
+            $message = $this->consumer->consume($this->config['consumer_timeout_ms']);
+        } catch (KafkaException) {
+            return null;
+        }
+
+        return match ($message->err) {
+            RD_KAFKA_RESP_ERR_NO_ERROR => $message,
+            RD_KAFKA_RESP_ERR__PARTITION_EOF, RD_KAFKA_RESP_ERR__TIMED_OUT, RD_KAFKA_RESP_ERR__UNKNOWN_PARTITION, RD_KAFKA_RESP_ERR__UNKNOWN_TOPIC, RD_KAFKA_RESP_ERR_UNKNOWN_TOPIC_OR_PART => null,
+            default => throw new RuntimeException($message->errstr(), $message->err),
+        };
+    }
+
+    /**
+     * @throws KafkaException
+     */
+    public function commit():void
+    {
+        $this->consumer->commit();
+    }
+
+    private function generateConfig(array $config): KafkaConfig
+    {
+        $kafkaConfig = new KafkaConfig();
+        $kafkaConfig->set('metadata.broker.list', $config['broker_list']);
+        $kafkaConfig->set('group.id', $config['group_name']);
+        $kafkaConfig->set('heartbeat.interval.ms', $config['heartbeat_ms']);
+        $kafkaConfig->set('auto.offset.reset', 'earliest');
+
+        return $kafkaConfig;
+    }
+
+    /**
+     * @throws KafkaException
+     */
+    private function checkSubscription(string $topic): void
+    {
+        if (!in_array($topic, $this->consumer->getSubscription())) {
+            $this->consumer->subscribe([$topic]);
+        }
+    }
+}

src/Kafka/Producer.php

@@ -0,0 +1,54 @@
+<?php
+
+namespace LaravelTool\KafkaQueue\Kafka;
+
+use Closure;
+use Illuminate\Support\Str;
+use RdKafka\Conf as KafkaConfig;
+use RdKafka\Producer as KafkaProducer;
+
+class Producer
+{
+    private KafkaProducer $producer;
+
+    public function __construct(
+        protected array $config
+    ) {
+        $this->producer = new KafkaProducer($this->generateConfig($config));
+    }
+
+    public function produce(
+        string $topic,
+        string $payload,
+        ?int $delay = null,
+        callable $callback = null,
+    ): void {
+        $this->producer->newTopic($topic)->producev(
+            partition: RD_KAFKA_PARTITION_UA,
+            msgflags: 0,
+            payload: $payload,
+            key: $this->getMessageKey(),
+            timestamp_ms: ($delay ?? time()) * 1000,
+        );
+
+        if (is_callable($callback)) {
+            $callback();
+        }
+
+        $this->producer->poll(0);
+        $this->producer->flush($this->config['producer_timeout_ms']);
+    }
+
+    private function getMessageKey(): string
+    {
+        return Str::orderedUuid()->toString();
+    }
+
+    private function generateConfig(array $config): KafkaConfig
+    {
+        $kafkaConfig = new KafkaConfig();
+        $kafkaConfig->set('metadata.broker.list', $config['broker_list']);
+
+        return $kafkaConfig;
+    }
+}