components/KafkaRequestReply.php - master - av_androsov/kafka-test

components/KafkaRequestReply.php
78 строк2 KB
Alexander Androsov
di kp
02 мар 2026, 12:04
02 мар 2026, 12:04abdfbdd
<?php
namespace common\components;

use Ramsey\Uuid\Uuid;
use Yii;

class KafkaRequestReply
{
    private KafkaProducer $producer;
    private string $requestTopic = 'gpt-requests';
    private string $responseTopic = 'gpt-responses';
    private int $timeout = 30000;

    public function __construct()
    {
        $this->producer = Yii::$container->get(KafkaProducer::class);
    }

    /**
     * @param array $data
     *
     * @return array|null
     */
    public function sendAndWait(array $data): ?array
    {
        $correlationId = Uuid::uuid4()->toString();

        $message = [
            'correlation_id' => $correlationId,
            'reply_to' => $this->responseTopic,
            'timestamp' => time(),
            'data' => $data
        ];

        if (!$this->producer->send($this->requestTopic, $message)) {
            return null;
        }

        return $this->waitForResponse($correlationId);
    }

    /**
     * @param string $correlationId
     *
     * @return array|null
     */
    private function waitForResponse(string $correlationId): ?array
    {
        $start = microtime(true);
        $timeoutSec = $this->timeout / 1000;

        $conf = new \RdKafka\Conf();
        $conf->set('bootstrap.servers', Yii::$app->params['kafka']['broker']);
        $conf->set('group.id', 'request-reply-' . uniqid());
        $conf->set('auto.offset.reset', 'earliest');
        $conf->set('enable.auto.commit', 'false');
        $conf->set('api.version.request', 'false');
        $conf->set('broker.version.fallback', '0.10.0.0');

        $consumer = new \RdKafka\KafkaConsumer($conf);
        $consumer->subscribe([$this->responseTopic]);

        while (microtime(true) - $start < $timeoutSec) {
            $message = $consumer->consume(1000);

            if ($message->err === RD_KAFKA_RESP_ERR_NO_ERROR) {
                $payload = json_decode($message->payload, true);
                $msgCorrelationId = $payload['correlation_id'] ?? null;

                if ($msgCorrelationId === $correlationId) {
                    return $payload['data'] ?? $payload;
                }
            }
        }

        return null;
    }
}