diswebru/laravel-kafka-tools

tools for mateusjunges/laravel-kafka

1.0.0 2025-03-05 15:42 UTC

Requires

• php: >=8.2
• mateusjunges/laravel-kafka: ^2.5

MIT

• Denis Romashkov

README

Installation

composer require diswebru/laravel-kafka-tools

Examples

Sending a message to the tests topic

use Diswebru\LaravelKafkaTools\Kafka;

Kafka::publish('topic', ['message-key' => 'message-value']);

Retrieve unprocessed messages from the tests topic and terminate the process

use Diswebru\LaravelKafkaTools\Kafka;

Kafka::consumer('topic', function (ConsumerMessage $message) {
        $data = $message->getBody();
        
        if (!isset($data['message-key']) && $data['message-key'] != 'message-value') {
            // There will be no commit
            throw new \Exception('Error message');
        }
    });

Retrieve unprocessed messages from the tests topic and terminate the process using mateusjunges/laravel-kafka

use Diswebru\LaravelKafkaTools\Infrastructure\Factories\ManuallyCommitterFactory;
use Junges\Kafka\Contracts\ConsumerMessage;
use Junges\Kafka\Facades\Kafka;

Kafka::consumer()
    ->subscribe('topic')
    ->withOptions([
        'enable.auto.commit' => 'false',
        'auto.offset.reset' => 'earliest'
    ])
    ->stopAfterLastMessage()
    ->usingCommitterFactory(new ManuallyCommitterFactory())
    ->withHandler(function (ConsumerMessage $message) {
        $data = $message->getBody();
        
        if (!isset($data['message-key']) && $data['message-key'] != 'message-value') {
            // There will be no commit
            throw new \Exception('Error message');
        }
    })
    ->build()
    ->consume();