RdKafkaCaster.php 4.5 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186
  1. <?php
  2. /*
  3. * This file is part of the Symfony package.
  4. *
  5. * (c) Fabien Potencier <fabien@symfony.com>
  6. *
  7. * For the full copyright and license information, please view the LICENSE
  8. * file that was distributed with this source code.
  9. */
  10. namespace Symfony\Component\VarDumper\Caster;
  11. use RdKafka\Conf;
  12. use RdKafka\Exception as RdKafkaException;
  13. use RdKafka\KafkaConsumer;
  14. use RdKafka\Message;
  15. use RdKafka\Metadata\Broker as BrokerMetadata;
  16. use RdKafka\Metadata\Collection as CollectionMetadata;
  17. use RdKafka\Metadata\Partition as PartitionMetadata;
  18. use RdKafka\Metadata\Topic as TopicMetadata;
  19. use RdKafka\Topic;
  20. use RdKafka\TopicConf;
  21. use RdKafka\TopicPartition;
  22. use Symfony\Component\VarDumper\Cloner\Stub;
  23. /**
  24. * Casts RdKafka related classes to array representation.
  25. *
  26. * @author Romain Neutron <imprec@gmail.com>
  27. */
  28. class RdKafkaCaster
  29. {
  30. public static function castKafkaConsumer(KafkaConsumer $c, array $a, Stub $stub, bool $isNested)
  31. {
  32. $prefix = Caster::PREFIX_VIRTUAL;
  33. try {
  34. $assignment = $c->getAssignment();
  35. } catch (RdKafkaException $e) {
  36. $assignment = [];
  37. }
  38. $a += [
  39. $prefix.'subscription' => $c->getSubscription(),
  40. $prefix.'assignment' => $assignment,
  41. ];
  42. $a += self::extractMetadata($c);
  43. return $a;
  44. }
  45. public static function castTopic(Topic $c, array $a, Stub $stub, bool $isNested)
  46. {
  47. $prefix = Caster::PREFIX_VIRTUAL;
  48. $a += [
  49. $prefix.'name' => $c->getName(),
  50. ];
  51. return $a;
  52. }
  53. public static function castTopicPartition(TopicPartition $c, array $a)
  54. {
  55. $prefix = Caster::PREFIX_VIRTUAL;
  56. $a += [
  57. $prefix.'offset' => $c->getOffset(),
  58. $prefix.'partition' => $c->getPartition(),
  59. $prefix.'topic' => $c->getTopic(),
  60. ];
  61. return $a;
  62. }
  63. public static function castMessage(Message $c, array $a, Stub $stub, bool $isNested)
  64. {
  65. $prefix = Caster::PREFIX_VIRTUAL;
  66. $a += [
  67. $prefix.'errstr' => $c->errstr(),
  68. ];
  69. return $a;
  70. }
  71. public static function castConf(Conf $c, array $a, Stub $stub, bool $isNested)
  72. {
  73. $prefix = Caster::PREFIX_VIRTUAL;
  74. foreach ($c->dump() as $key => $value) {
  75. $a[$prefix.$key] = $value;
  76. }
  77. return $a;
  78. }
  79. public static function castTopicConf(TopicConf $c, array $a, Stub $stub, bool $isNested)
  80. {
  81. $prefix = Caster::PREFIX_VIRTUAL;
  82. foreach ($c->dump() as $key => $value) {
  83. $a[$prefix.$key] = $value;
  84. }
  85. return $a;
  86. }
  87. public static function castRdKafka(\RdKafka $c, array $a, Stub $stub, bool $isNested)
  88. {
  89. $prefix = Caster::PREFIX_VIRTUAL;
  90. $a += [
  91. $prefix.'out_q_len' => $c->getOutQLen(),
  92. ];
  93. $a += self::extractMetadata($c);
  94. return $a;
  95. }
  96. public static function castCollectionMetadata(CollectionMetadata $c, array $a, Stub $stub, bool $isNested)
  97. {
  98. $a += iterator_to_array($c);
  99. return $a;
  100. }
  101. public static function castTopicMetadata(TopicMetadata $c, array $a, Stub $stub, bool $isNested)
  102. {
  103. $prefix = Caster::PREFIX_VIRTUAL;
  104. $a += [
  105. $prefix.'name' => $c->getTopic(),
  106. $prefix.'partitions' => $c->getPartitions(),
  107. ];
  108. return $a;
  109. }
  110. public static function castPartitionMetadata(PartitionMetadata $c, array $a, Stub $stub, bool $isNested)
  111. {
  112. $prefix = Caster::PREFIX_VIRTUAL;
  113. $a += [
  114. $prefix.'id' => $c->getId(),
  115. $prefix.'err' => $c->getErr(),
  116. $prefix.'leader' => $c->getLeader(),
  117. ];
  118. return $a;
  119. }
  120. public static function castBrokerMetadata(BrokerMetadata $c, array $a, Stub $stub, bool $isNested)
  121. {
  122. $prefix = Caster::PREFIX_VIRTUAL;
  123. $a += [
  124. $prefix.'id' => $c->getId(),
  125. $prefix.'host' => $c->getHost(),
  126. $prefix.'port' => $c->getPort(),
  127. ];
  128. return $a;
  129. }
  130. private static function extractMetadata($c)
  131. {
  132. $prefix = Caster::PREFIX_VIRTUAL;
  133. try {
  134. $m = $c->getMetadata(true, null, 500);
  135. } catch (RdKafkaException $e) {
  136. return [];
  137. }
  138. return [
  139. $prefix.'orig_broker_id' => $m->getOrigBrokerId(),
  140. $prefix.'orig_broker_name' => $m->getOrigBrokerName(),
  141. $prefix.'brokers' => $m->getBrokers(),
  142. $prefix.'topics' => $m->getTopics(),
  143. ];
  144. }
  145. }