README

This package can be installed as a Composer dependency.

composer require kim1ne/loop-server

This component unites all libraries, where needs start an infinite process. Example:

Producers publishing messages to the kafka. Workers expect messages from the kafka and when got messages, sending to the socket server. The socket-server sending messages to all connected clients

use RdKafka\Conf;
use Kim1ne\Socket\Server\Server;
use Kim1ne\Socket\Server\Connection;
use Kim1ne\Socket\Server\Message;
use Kim1ne\Socket\Server\Transport;
use Kim1ne\Kafka\KafkaWorker;
use Kim1ne\Kafka\KafkaConsumer;
use Kim1ne\Kafka\Message as KafkaMessage;

$server = $server = new Server(transport: Transport::WS);

$server->on('message', function (Message $message, Connection $connection, Server $server) {
    InputMessage::green("I've got the message!");
});

$conf = new Conf(...);

$worker = new KafkaWorker($conf);
$worker->subscribe(['my-topic'])

$worker->on(function (KafkaMessage $message, KafkaConsumer $consumer) use ($server) {
    $consumer->commitAsync($message);
    $server->sendAll($message->payload);
});

\Kim1ne\Loop\Server::run($server, $worker);

use Kim1ne\Socket\Server\Server;
use Kim1ne\Kafka\KafkaWorker;
use Kim1ne\Kafka\Message;
use Kim1ne\Kafka\KafkaConsumer;
use Kim1ne\Core\Event;

/**
 * @var Server $server
 */
$server->on('kafka:worker:message', function (Event $event) use ($server) {
    $message = $event->get('message');
    $server->sendAll($message);
});

/**
 * @var KafkaWorker $worker
 */
$worker->on('message', function (Message $message, KafkaConsumer $consumer) use ($worker) {
    $worker->dispatchEvent('message', new Event([
        'message' => $message->payload
    ]));
});