<?php
namespace app\common\library;

use RdKafka\Conf;
use RdKafka\Producer;
use RdKafka\KafkaConsumer;
use think\Config;

class Kafka
{
    private $config;
    private static $instance = null;

    private function __construct()
    {
        $this->config = Config::get('kafka.');
    }

    public static function getInstance()
    {
        if (self::$instance === null) {
            self::$instance = new self();
        }
        return self::$instance;
    }

    /**
     * 获取生产者实例
     */
    public function getProducer()
    {
        $conf = new Conf();
        $conf->set('metadata.broker.list', $this->config['bootstrap_servers']);
        
        if ($this->config['debug']) {
            $conf->set('debug', 'all');
        }

        return new Producer($conf);
    }

    /**
     * 获取消费者实例
     */
    public function getConsumer()
    {
        $conf = new Conf();
        $conf->set('metadata.broker.list', $this->config['bootstrap_servers']);
        $conf->set('group.id', $this->config['group_id']);
        $conf->set('auto.offset.reset', 'earliest');

        if ($this->config['debug']) {
            $conf->set('debug', 'all');
        }

        return new KafkaConsumer($conf);
    }

    /**
     * 发送消息
     */
    public function publish($topic, $message, $key = null)
    {
        $producer = $this->getProducer();
        $topic = $producer->newTopic($topic);

        $topic->produce(RD_KAFKA_PARTITION_UA, 0, json_encode($message), $key);
        $producer->flush(10000);
    }

    /**
     * 消费消息
     */
    public function consume($topics, callable $callback)
    {
        $consumer = $this->getConsumer();
        $consumer->subscribe($topics);

        while (true) {
            $message = $consumer->consume(120 * 1000);
            switch ($message->err) {
                case RD_KAFKA_RESP_ERR_NO_ERROR:
                    $callback($message);
                    break;
                case RD_KAFKA_RESP_ERR__PARTITION_EOF:
                    echo "No more messages; will wait...\n";
                    break;
                case RD_KAFKA_RESP_ERR__TIMED_OUT:
                    echo "Timed out\n";
                    break;
                default:
                    throw new \Exception($message->errstr(), $message->err);
            }
        }
    }
}