kiri-core/Kafka/Producer.php

<?php
declare(strict_types=1);

namespace Kafka;

use Exception;
use RdKafka\Conf;
use RdKafka\ProducerTopic;
use RdKafka\TopicConf;
use Snowflake\Abstracts\Component;
use Snowflake\Abstracts\Config;
use Snowflake\Exception\ConfigException;
use Snowflake\Snowflake;

/**
 * +------------------------------------------------------------------------------
 * Kafka protocol since Kafka v0.8
 * +------------------------------------------------------------------------------
 *
 * @package
 * @version $_SWANBR_VERSION_$
 * @copyright Copyleft
 * @author $_SWANBR_AUTHOR_$
 * +------------------------------------------------------------------------------
 */
class Producer extends Component
{


	private Conf $conf;
	private TopicConf $topicConf;

	private ?\RdKafka\Producer $producer = null;

	private bool $isAck = true;

	/**
	 * Producer constructor.
	 * @param array $config
	 * @throws Exception
	 */
	public function __construct(array $config = [])
	{
		parent::__construct($config);
		if (!class_exists(Conf::class)) {
			return;
		}
		$this->conf = di(Conf::class);
		$this->conf->setErrorCb(function ($kafka, $err, $reason) {
			$this->error(sprintf("Kafka error: %s (reason: %s)", rd_kafka_err2str($err), $reason));
		});
		$this->topicConf = di(TopicConf::class);
	}


	/**
	 * @param string $servers
	 * @return Producer
	 */
	public function setBrokers(string $servers): static
	{
		$this->conf->set('metadata.broker.list', $servers);
		return $this;
	}


	/**
	 * @param string $groupId
	 * @return Producer
	 */
	public function setGroupId(string $groupId): static
	{
		$this->conf->set('group.id', $groupId);
		return $this;
	}


	/**
	 * @param string $topic
	 * @param array $params
	 * @param string|null $groupId
	 * @throws Exception
	 */
	public function dispatch(string $topic, array $params = [], string $groupId = null)
	{
		$this->beforePushMessage($topic, $groupId);
		$this->sendMessage($topic, [$params]);
	}

	/**
	 * @return \RdKafka\Producer
	 * @throws Exception
	 */
	private function getProducer(): \RdKafka\Producer
	{
		return Snowflake::getDi()->get(\RdKafka\Producer::class, [$this->conf]);
	}


	/**
	 * @param \RdKafka\Producer $producer
	 * @param $topic
	 * @return ProducerTopic
	 * @throws Exception
	 */
	private function getProducerTopic(\RdKafka\Producer $producer, $topic): ProducerTopic
	{
		return $producer->newTopic($topic, $this->topicConf);
	}


	/**
	 * @param string $toPic
	 * @param string|null $key
	 * @param array $data
	 * @param string|null $groupId
	 * @throws ConfigException
	 * @throws Exception
	 */
	public function batch(string $toPic, ?string $key, array $data, ?string $groupId = null)
	{
		$this->beforePushMessage($toPic, $groupId);

		$this->sendMessage($toPic, $data, $key);
	}


	/**
	 * @param $topic
	 * @param $groupId
	 * @throws ConfigException
	 * @throws Exception
	 */
	private function beforePushMessage($topic, $groupId): void
	{
		$consumers = Config::get('kafka.producers.' . $topic);
		if (empty($consumers) || !is_array($consumers)) {
			throw new Exception('You need set kafka.producers config');
		}
		if (!isset($consumers['brokers'])) {
			throw new Exception('You need set brokers config.');
		}
		if (!empty($groupId)) {
			$consumers['groupId'] = $groupId;
		} else if (!isset($consumers['groupId'])) {
			$consumers['groupId'] = $topic . ':' . Snowflake::localhost();
		}
		$this->setGroupId($consumers['groupId']);
		$this->setBrokers($consumers['brokers']);
	}


	/**
	 * @param string $topic
	 * @param array $message
	 * @param string $key
	 * @throws Exception
	 */
	private function sendMessage(string $topic, array $message, string $key = '')
	{
		$producer = $this->getProducer();
		$producerTopic = $this->getProducerTopic($producer, $topic);
		if ($this->isAck) {
			$this->flush($producer);
		}
		foreach ($message as $value) {
			$producerTopic->produce(RD_KAFKA_PARTITION_UA, 0, swoole_serialize($value), $key);
			$producer->poll(0);
		}
		$this->flush($producer);
	}


	/**
	 * @param bool $ack
	 */
	public function setAsk(bool $ack)
	{
		$this->isAck = $ack;
		$this->topicConf->set('request.required.acks', $this->isAck ? '1' : '0');
	}


	/**
	 * @param \RdKafka\Producer $producer
	 */
	public function flush(\RdKafka\Producer $producer)
	{
		while ($producer->getOutQLen() > 0) {
			$result = $producer->flush(100);
			if (RD_KAFKA_RESP_ERR_NO_ERROR === $result) {
				break;
			}
		}
	}

}