<commit_before><?php

	namespace vps\tools\components;

	use vps\tools\helpers\Console;
	use Yii;
	use yii\base\Component;
	use yii\helpers\Json;
	use yii\web\UnprocessableEntityHttpException;

	class KafkaRest extends Component
	{
		/**
		 * The kafka host.
		 *
		 * @var string|null
		 */
		public $host;
		/**
		 * The kafka port.
		 *
		 * @var string|null
		 */
		public $port;
		/**
		 * Name topic.
		 *
		 * @var string|null
		 */
		public $source;
		/**
		 * Name topic.
		 *
		 * @var string|null
		 */
		public $topic;
		/**
		 * Use kafka.
		 *
		 * @var string|null
		 */
		public $use;

		/**
		 * @inheritdoc
		 */
		public function init ()
		{
			parent::init();

			$this->host = Yii::$app->settings->get('kafka_rest_host');
			$this->port = Yii::$app->settings->get('kafka_rest_port');
			$this->topic = Yii::$app->settings->get('kafka_topic');
			$this->source = Yii::$app->settings->get('kafka_source');
			$this->use = Yii::$app->settings->get('kafka_use');
		}

		/**
		 * @inheritdoc
		 */
		public function sendMessage ($data)
		{
			if ($this->use)
			{
				try
				{
					$records[ 'records' ][][ 'value' ] = $data;

					$ch = curl_init();
					curl_setopt($ch, CURLOPT_URL, $this->host . ':' . $this->port . '/topics/' . $this->source);
					curl_setopt($ch, CURLOPT_POST, 1);
					curl_setopt($ch, CURLOPT_POSTFIELDS, Json::encode($records));
					curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);
					$headers = [
						'Content-Type:application/vnd.kafka.json.v2+json'
					];

					curl_setopt($ch, CURLOPT_HTTPHEADER, $headers);

					$server_output = curl_exec($ch);

					curl_close($ch);
					$responce = Json::decode($server_output);
				
					if (isset($responce[ 'offsets' ][ 0 ][ 'offset' ]) and $responce[ 'offsets' ][ 0 ][ 'offset' ] > 0)
					{
						if (Yii::$app->has('logging'))
							Yii::$app->logging->info(Yii::tr('Данные для {object} отправленны  в Kafka.', [ 'object' => json_encode($data[ 'id' ]) ]));

						return true;
					}
					else
					{
						if (Yii::$app->has('logging'))
							Yii::$app->logging->error(Yii::tr('Ошибка {error} отправки сообщения в kafka.' . Json::encode($data), [ 'error' => isset($responce[ 'offsets' ][ 0 ][ 'error_code' ]) ? $responce[ 'offsets' ][ 0 ][ 'error_code' ] . ' ' . $responce[ 'offsets' ][ 0 ][ 'error' ] : '' ]));

						return false;
					}
				}
				catch (\Exception $e)
				{
					if (YII_DEBUG)
						throw new UnprocessableEntityHttpException($e->getMessage() . ' ' . $e->getLine() . ' ' . $e->getTraceAsString());
					Yii::error($e->getMessage());

					return null;
				}
			}
		}

		public function initConsumer ()
		{
			try
			{
				$records[ 'name' ] = $this->topic;
				$records[ 'format' ] = 'json';
				$records[ 'auto.offset.reset' ] = 'earliest';
				$ch = curl_init();
				curl_setopt($ch, CURLOPT_URL, $this->host . ':' . $this->port . '/consumers/' . $this->topic);
				curl_setopt($ch, CURLOPT_POST, 1);
				curl_setopt($ch, CURLOPT_POSTFIELDS, Json::encode($records));
				curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);

				$headers = [
					'Content-Type:application/vnd.kafka.v2+json'
				];

				curl_setopt($ch, CURLOPT_HTTPHEADER, $headers);

				$server_output = curl_exec($ch);

				curl_close($ch);
				$responce = Json::decode($server_output);

				$data[ 'topics' ] = [ $this->topic ];
				$ch = curl_init();
				curl_setopt($ch, CURLOPT_URL, $this->host . ':' . $this->port . '/consumers/' . $this->topic . '/instances/' . $this->topic . '/subscription');
				curl_setopt($ch, CURLOPT_POST, 1);
				curl_setopt($ch, CURLOPT_POSTFIELDS, Json::encode($data));
				curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);

				$headers = [
					'Content-Type:application/vnd.kafka.v2+json'
				];

				curl_setopt($ch, CURLOPT_HTTPHEADER, $headers);

				$server_output = curl_exec($ch);

				curl_close($ch);
				$responce = Json::decode($server_output);
			}<commit_msg> some comments<commit_after><?php

	namespace vps\tools\components;

	use vps\tools\helpers\Console;
	use Yii;
	use yii\base\Component;
	use yii\helpers\Json;
	use yii\web\UnprocessableEntityHttpException;

	class KafkaRest extends Component
	{
		/**
		 * The kafka host.
		 *
		 * @var string|null
		 */
		public $host;
		/**
		 * The kafka port.
		 *
		 * @var string|null
		 */
		public $port;
		/**
		 * Name topic.
		 *
		 * @var string|null
		 */
		public $source;
		/**
		 * Name topic.
		 *
		 * @var string|null
		 */
		public $topic;
		/**
		 * Use kafka.
		 *
		 * @var string|null
		 */
		public $use;

		/**
		 * @inheritdoc
		 */
		public function init ()
		{
			parent::init();

			$this->host = Yii::$app->settings->get('kafka_rest_host');
			$this->port = Yii::$app->settings->get('kafka_rest_port');
			$this->topic = Yii::$app->settings->get('kafka_topic');
			$this->source = Yii::$app->settings->get('kafka_source');
			$this->use = Yii::$app->settings->get('kafka_use');
		}

		/**
		 * @inheritdoc
		 */
		public function sendMessage ($data)
		{
			if ($this->use)
			{
				try
				{
					$records[ 'records' ][][ 'value' ] = $data;

					// Не через CurlTransport, потому не даёт выставить заголовок Content-Type:application/vnd.kafka.json.v2+json
					// Меняет автоматом на стандартные form или json.
					$ch = curl_init();
					curl_setopt($ch, CURLOPT_URL, $this->host . ':' . $this->port . '/topics/' . $this->source);
					curl_setopt($ch, CURLOPT_POST, 1);
					curl_setopt($ch, CURLOPT_POSTFIELDS, Json::encode($records));
					curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);
					$headers = [
						'Content-Type:application/vnd.kafka.json.v2+json'
					];

					curl_setopt($ch, CURLOPT_HTTPHEADER, $headers);

					$server_output = curl_exec($ch);

					curl_close($ch);
					$responce = Json::decode($server_output);

					if (isset($responce[ 'offsets' ][ 0 ][ 'offset' ]) and $responce[ 'offsets' ][ 0 ][ 'offset' ] > 0)
					{
						if (Yii::$app->has('logging'))
							Yii::$app->logging->info(Yii::tr('Данные для {object} отправленны  в Kafka.', [ 'object' => json_encode($data[ 'id' ]) ]));

						return true;
					}
					else
					{
						if (Yii::$app->has('logging'))
							Yii::$app->logging->error(Yii::tr('Ошибка {error} отправки сообщения в kafka.' . Json::encode($data), [ 'error' => isset($responce[ 'offsets' ][ 0 ][ 'error_code' ]) ? $responce[ 'offsets' ][ 0 ][ 'error_code' ] . ' ' . $responce[ 'offsets' ][ 0 ][ 'error' ] : '' ]));

						return false;
					}
				}
				catch (\Exception $e)
				{
					if (YII_DEBUG)
						throw new UnprocessableEntityHttpException($e->getMessage() . ' ' . $e->getLine() . ' ' . $e->getTraceAsString());
					Yii::error($e->getMessage());

					return null;
				}
			}

			return false;
		}

		public function initConsumer ()
		{
			try
			{
				$records[ 'name' ] = $this->topic;
				$records[ 'format' ] = 'json';
				$records[ 'auto.offset.reset' ] = 'earliest';
				$ch = curl_init();
				curl_setopt($ch, CURLOPT_URL, $this->host . ':' . $this->port . '/consumers/' . $this->topic);
				curl_setopt($ch, CURLOPT_POST, 1);
				curl_setopt($ch, CURLOPT_POSTFIELDS, Json::encode($records));
				curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);

				$headers = [
					'Content-Type:application/vnd.kafka.v2+json'
				];

				curl_setopt($ch, CURLOPT_HTTPHEADER, $headers);

				$server_output = curl_exec($ch);

				curl_close($ch);
				$response = Json::decode($server_output);

				$data[ 'topics' ] = [ $this->topic ];
				$ch = curl_init();
				curl_setopt($ch, CURLOPT_URL, $this->host . ':' . $this->port . '/consumers/' . $this->topic . '/instances/' . $this->topic . '/subscription');
				curl_setopt($ch, CURLOPT_POST, 1);
				curl_setopt($ch, CURLOPT_POSTFIELDS, Json::encode($data));
				curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);

				$headers = [
					'Content-Type:application/vnd.kafka.v2+json'
				];

				curl_setopt($ch, CURLOPT_HTTPHEADER, $headers);

				$server_output = curl_exec($ch);

				curl_close($ch);
				$response = Json::decode($server_output);
			}
{'fraction_non_alphanumeric': 0.19335423197492163, 'fraction_numerical': 0.003134796238244514, 'mean_word_length': 2.9979949874686715, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 85, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '748653', 'n_tokens_mistral': 3690, 'n_tokens_neox': 2910, 'n_words': 496}