* * For the full copyright and license information, please view the LICENSE * file that was distributed with this source code. */ namespace Symfony\Component\VarDumper\Tests\Caster; use PHPUnit\Framework\TestCase; use RdKafka\Conf; use RdKafka\KafkaConsumer; use RdKafka\Producer; use RdKafka\TopicConf; use Symfony\Component\VarDumper\Test\VarDumperTestTrait; /** * @requires extension rdkafka * @group integration */ class RdKafkaCasterTest extends TestCase { use VarDumperTestTrait; private const TOPIC = 'test-topic'; private const GROUP_ID = 'test-group-id'; private $hasBroker = false; private $broker; protected function setUp(): void { if (!$this->hasBroker && getenv('KAFKA_BROKER')) { $this->broker = getenv('KAFKA_BROKER'); $this->hasBroker = true; } } public function testDumpConf() { $conf = new Conf(); $conf->setErrorCb(function ($kafka, $err, $reason) {}); $conf->setDrMsgCb(function () {}); $conf->setRebalanceCb(function () {}); // BC with earlier version of extension rdkafka foreach (['setLogCb', 'setOffsetCommitCb', 'setStatsCb', 'setConsumeCb'] as $method) { if (method_exists($conf, $method)) { $conf->{$method}(function () {}); } } $expectedDump = <<assertDumpMatchesFormat($expectedDump, $conf); } public function testDumpProducer() { if (!$this->hasBroker) { $this->markTestSkipped('Test requires an active broker'); } $producer = new Producer(new Conf()); $producer->addBrokers($this->broker); $expectedDump = <<broker/1001" brokers: RdKafka\Metadata\Collection { +0: RdKafka\Metadata\Broker { id: 1001 host: "%s" port: %d } } topics: RdKafka\Metadata\Collection { +0: RdKafka\Metadata\Topic { name: "%s" partitions: RdKafka\Metadata\Collection { +0: RdKafka\Metadata\Partition { id: 0 err: 0 leader: 1001 }%A } }%A } } EODUMP; $this->assertDumpMatchesFormat($expectedDump, $producer); } public function testDumpTopicConf() { $topicConf = new TopicConf(); $topicConf->set('auto.offset.reset', 'smallest'); $expectedDump = <<assertDumpMatchesFormat($expectedDump, $topicConf); } public function testDumpKafkaConsumer() { if (!$this->hasBroker) { $this->markTestSkipped('Test requires an active broker'); } $conf = new Conf(); $conf->set('metadata.broker.list', $this->broker); $conf->set('group.id', self::GROUP_ID); $consumer = new KafkaConsumer($conf); $consumer->subscribe([self::TOPIC]); $expectedDump = << "test-topic" ] assignment: [] orig_broker_id: %i orig_broker_name: "$this->broker/%s" brokers: RdKafka\Metadata\Collection { +0: RdKafka\Metadata\Broker { id: 1001 host: "%s" port: %d } } topics: RdKafka\Metadata\Collection { +0: RdKafka\Metadata\Topic { name: "%s" partitions: RdKafka\Metadata\Collection { +0: RdKafka\Metadata\Partition { id: 0 err: 0 leader: 1001 }%A } }%A } } EODUMP; $this->assertDumpMatchesFormat($expectedDump, $consumer); } public function testDumpProducerTopic() { $producer = new Producer(new Conf()); $producer->addBrokers($this->broker); $topic = $producer->newTopic('test'); $topic->produce(RD_KAFKA_PARTITION_UA, 0, '{}'); $expectedDump = <<assertDumpMatchesFormat($expectedDump, $topic); } public function testDumpMessage() { $conf = new Conf(); $conf->set('metadata.broker.list', $this->broker); $conf->set('group.id', self::GROUP_ID); $consumer = new KafkaConsumer($conf); $consumer->subscribe([self::TOPIC]); // Force timeout $message = $consumer->consume(0); $expectedDump = <<assertDumpMatchesFormat($expectedDump, $message); } }