add more logs

Author: danikdantist

composer.json

@@ -1,6 +1,6 @@
 {
   "name": "danikdantist/queue-wrapper",
-  "version": "0.0.3",
+  "version": "0.0.4",
   "description": "My first Composer project",
   "keywords": [
     "queue",

src/Drivers/Kafka/Consumer.php

@@ -75,6 +75,10 @@ protected function init()
         $conf->set("enable.auto.commit", "false");
         $conf->set("enable.auto.offset.store", "false");
 
+        $conf->setErrorCb(function ($kafka, $err, $reason) {
+            $this->logError(sprintf("%s (reason: %s)\n", rd_kafka_err2str($err), $reason));
+        });
+
         $topicConf = new \RdKafka\TopicConf();
 
         $topicConf->set('auto.commit.enable', 'false');

src/Drivers/Kafka/Producer.php

@@ -51,8 +51,10 @@ protected function init()
         $conf->set('queued.max.messages.kbytes',100000000);
         $conf->set('socket.send.buffer.bytes',1000000);
         $conf->set('queue.buffering.max.messages',10000000);
-        //$conf->set('queue.buffering.max.ms',1);
 
+        $conf->setErrorCb(function ($kafka, $err, $reason) {
+            $this->logError(sprintf("%s (reason: %s)\n", rd_kafka_err2str($err), $reason));
+        });
 
         $rk = new \RdKafka\Producer($conf);
         $rk->setLogLevel(LOG_DEBUG);