<?php

namespace app\library;

/**
 * Kafka操作类
 */
trait KafkaClient
{
    /**
     * 定义Kafka生产者初始化配置
     * 需要在服务端配置 kafka_server_jaas.conf 里面添加对应的用户名和密码
     * @var array
     */
    private $kafkaProducerConfig =  [];

    /**
     * 定义Kafka生产者初始化配置
     * 需要在服务端配置 kafka_client_jaas.conf 里面添加对应的用户名和密码
     * @var array
     */
    private $kafkaConsumerConfig =  [];


    /**
     * 初始化kafka生产配置
     * @return string[]
     */
    private function initKafkaProducerConfig() :array
    {
        return $this->kafkaProducerConfig = [
            'username'      => 'orico',
            'password'      => 'password',/**/
            'broker_list'   => '127.0.0.1:9092',/*如果有多个，ip:port,ip2:port,ip3:port*/
            'topic'         => 'topics'
        ];
    }

    /**
     * 初始化kafka消费配置
     * @return string[]
     */
    private function initKafkaConsumerConfig() :array
    {
        return $this->kafkaConsumerConfig = [
            'username'      => 'orico',
            'password'      => 'password',/**/
            'brokerlist'    => '127.0.0.1:9092',/*如果有多个，ip:port,ip2:port,ip3:port*/
            'topic'         => 'topics'
        ];
    }

    /**
     * kafka 生产者核心方法
     * @param array $config kafka 配置参数
     * @param array $data 需要生产的数据
     */
    public function kafkaProducer(array $config, array $data)
    {
        if (empty($this->kafkaProducerConfig)) {
            $this->initKafkaProducerConfig();
        }
        $conf = new \RdKafka\Conf();
        $conf->set('sasl.mechanisms', 'PLAIN');
        $conf->set('security.protocol', 'SASL_PLAINTEXT');
        $conf->set('sasl.username', $config['username']);
        $conf->set('sasl.password', $config['password']);
        $conf->set('socket.keepalive.enable', 'true'); // socket不操作保持长连
        $conf->set('log.connection.close', 'false'); // 静默所有自发断开连接日志
        $conf->set('metadata.broker.list', $config['broker_list']);

        $producer = new \RdKafka\Producer($conf);

        $topic = $producer->newTopic($this->topicProduce);

        for ($i = 0; $i < 10; $i++) {
            $topic->produce(RD_KAFKA_PARTITION_UA, 0,$data);
            $producer->poll(0);
        }

        for ($flushRetries = 0; $flushRetries < 10; $flushRetries++) {
            $result = $producer->flush(10000);
            if (RD_KAFKA_RESP_ERR_NO_ERROR === $result) {
                break;
            }
        }

        if (RD_KAFKA_RESP_ERR_NO_ERROR !== $result) {
            throw new \RuntimeException('Was unable to flush, messages might be lost!');
        }

    }


    /**
     * kafka 消息者核心方法
     * @param array $config kafka 配置参数
     * @param array $data 需要生产的数据
     */
    public function kafkaConsumer(array $config, array $data)
    {
        if (empty($this->kafkaConsumerConfig)) {
            $this->initKafkaConsumerConfig();
        }
        return ($this->kafkaConfig());
    }

    /**
     * curl请求
     * @param $url  string 请求的url链接
     * @param $data string|array|mixed 请求的数据
     * @param bool $is_post 是否是post请求，默认false
     * @param array $options 是否附带请求头
     * @return array|mixed
     */
    private function http(string $url, array $data, bool $is_post=false, array $options=[]):array
    {
        $data  = json_encode($data);
        $headerArray = [
            'Content-type: application/json;charset=utf-8',
            'Accept: application/json'
        ];
        $curl = curl_init();
        $arr = [];
        array_push($arr,$url);
        curl_setopt($curl, CURLOPT_URL, $url);
        curl_setopt($curl, CURLOPT_SSL_VERIFYPEER, FALSE);
        curl_setopt($curl, CURLOPT_SSL_VERIFYHOST,FALSE);
        if ($is_post) {
            curl_setopt($curl, CURLOPT_POST, 1);
            curl_setopt($curl, CURLOPT_POSTFIELDS, $data);
        }
        curl_setopt($curl,CURLOPT_HTTPHEADER,$headerArray);
        curl_setopt($curl, CURLOPT_RETURNTRANSFER, 1);
        if (!empty($options['cookie'])) {
            curl_setopt($curl, CURLOPT_COOKIE, $options['cookie']);
        }
        $output = curl_exec($curl);
        $http_status = curl_errno($curl);
        $http_msg = curl_error($curl);
        curl_close($curl);
        if ($http_status == 0) {
            return json_decode($output, true);
        } else {
            return ['status' => $http_status, 'message' => $http_msg, 'data' => []];
        }
    }


}