PHP code example of ensi / laravel-phprdkafka-consumer
1. Go to this page and download the library: Download ensi/laravel-phprdkafka-consumer library. Choose the download type require.
2. Extract the ZIP file and open the index.php.
3. Add this code to the index.php.
<?php
require_once('vendor/autoload.php');
/* Start to develop here. Best regards https://php-download.com/ */
ensi / laravel-phprdkafka-consumer example snippets
[
/*
| Optional, defaults to `null`.
| Here you may specify which topic should be handled by this processor.
| Processor handles all topics by default.
*/
'topic' => 'stage.crm.fact.registrations.1',
/*
| Optional, defaults to `null`.
| Here you may specify which ensi/laravel-phprdkafka consumer should be handled by this processor.
| Processor handles all consumers by default.
*/
'consumer' => 'default',
/*
| Optional, defaults to `action`.
| Here you may specify processor's type. Defaults to `action`
| Supported types:
| - `action` - a simple class with execute method;
| - `job` - Laravel Queue Job. It will be dispatched using `dispatch` or `dispatchSync` method;
*/
'type' => 'action',
/*
| Required.
| Fully qualified class name of a processor class.
*/
'class' => \App\Domain\Communication\Actions\SendConfirmationEmailAction::class,
/*
| Optional, defaults to `false`.
| Proxy messages to Laravel's queue.
| Supported values:
| - `false` - do not stream message. Execute processor in syncronous mode;
| - `true` - stream message to Laravel's default queue;
| - `<your-favorite-queue-name-as-string>` - stream message to this queue;
*/
'queue' => false,
/*
| Optional, defaults to 5000.
| Kafka consume timeout in milliseconds .
*/
'consume_timeout' => 5000,
]
use RdKafka\Message;
class SendConfirmationEmailAction
{
public function execute(Message $message): void
{
// var_dump($message->payload);
}
}
use RdKafka\Message;
use Illuminate\Bus\Queueable;
use Illuminate\Contracts\Queue\ShouldQueue;
use Illuminate\Foundation\Bus\Dispatchable;
use Illuminate\Queue\InteractsWithQueue;
class ConsumeMessageJob implements ShouldQueue
{
use Dispatchable, InteractsWithQueue, Queueable;
public function __construct(protected Message $message)
{
}
public function handle(): void
{
// var_dump($this->message->payload);
}
}