kiri-core/Kafka/Producer.php

<?php
declare(strict_types=1);

namespace Kafka;

use RdKafka\Conf;
use RdKafka\TopicConf;
use ReflectionException;
use Snowflake\Abstracts\Component;
use Snowflake\Event;
use Snowflake\Exception\NotFindClassException;
use Snowflake\Snowflake;

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

	private string $_topic = '';


	private Conf $conf;
	private TopicConf $topicConf;


	public function __construct($config = [])
	{
		parent::__construct($config);
		if (!class_exists(Conf::class)) {
			return;
		}
		$this->conf = new Conf();
		$this->topicConf = new TopicConf();
	}


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


	/**
	 * @param bool $value
	 * @return $this
	 */
	public function setAck(bool $value)
	{
		$this->topicConf->set('request.required.acks', (string)$value);
		return $this;
	}


	/**
	 * @param $servers
	 * @return Producer
	 */
	public function setTopic(string $servers)
	{
		$this->_topic = $servers;
		return $this;
	}


	/**
	 * @param $message
	 * @param null $key
	 * @param int $timeout
	 * @throws
	 */
	public function delivery($message, $key = null, $timeout = 10)
	{
		if (!$this->conf || !$this->topicConf) {
			throw new \Exception('Error. Please set kafka conf.');
		}
		$this->conf->setErrorCb(function ($kafka, $err, $reason) {
			$this->error(sprintf("Kafka error: %s (reason: %s)", rd_kafka_err2str($err), $reason));
		});

		/** @var \RdKafka\Producer $rk */
		$rk = Snowflake::createObject(\RdKafka\Producer::class, [$this->conf]);
		if ($rk->getOutQLen() > 0) {
			$this->clean($rk, $timeout);
		}
		$topic = $rk->newTopic($this->_topic, $this->topicConf);
		$topic->produce(RD_KAFKA_PARTITION_UA, 0, $message, $key);
		$this->clean($rk, $timeout);
	}


	/**
	 * @param $rk
	 * @param $timeout
	 */
	private function clean($rk, $timeout)
	{
		for ($length = 0; $length < $rk->getOutQLen(); $length++) {
			$rk->poll(0);
		}
		$rk->flush($timeout);
	}

}