<?php

/**
 * KafkaConfig (PHP 7.0-compatible variant)
 *
 * Identical in behaviour to ../KafkaConfig.php, but written so it parses and
 * runs on PHP 7.0+ (no typed properties, no ?type / void / mixed syntax).
 * Types are documented in docblocks instead.
 *
 * Central place for broker connection details and shared rdkafka
 * configuration. Keeping this in one class means the Producer and the
 * Consumer always agree on brokers/timeouts, and callers only have to
 * pass the pieces that actually change (topic, group, etc.).
 *
 * Requires the php-rdkafka extension (pecl install rdkafka).
 */
class KafkaConfig
{
    /** @var string Comma separated list of host:port brokers. */
    private $brokers;

    /** @var array<string,string> Extra librdkafka properties to merge in. */
    private $extra;

    /**
     * @param string               $brokers e.g. "127.0.0.1:9092" or "b1:9092,b2:9092"
     * @param array<string,string> $extra   any additional librdkafka config
     *                                       (e.g. security.protocol, sasl.*)
     */
    public function __construct($brokers = '127.0.0.1:9092', array $extra = [])
    {
        if ($brokers === '') {
            throw new \InvalidArgumentException('Kafka brokers list cannot be empty.');
        }
        $this->brokers = $brokers;
        $this->extra   = $extra;
    }

    /**
     * @return string
     */
    public function getBrokers()
    {
        return $this->brokers;
    }

    /**
     * Build a base \RdKafka\Conf pre-filled with the shared settings.
     * Both producer and consumer start from this.
     *
     * @return \RdKafka\Conf
     */
    public function baseConf()
    {
        $conf = new \RdKafka\Conf();
        $conf->set('metadata.broker.list', $this->brokers);
        // Log connection errors instead of dying silently.
        $conf->set('log_level', (string) LOG_WARNING);

        foreach ($this->extra as $key => $value) {
            $conf->set($key, (string) $value);
        }

        return $conf;
    }
}