Merge pull request #6 from Sin30/all-fix

Fix several bugs that stopping this queue from working #6 by Sin30

Author: petermein

src/Queue/Jobs/KafkaJob.php

@@ -3,6 +3,7 @@
 namespace Rapide\LaravelQueueKafka\Queue\Jobs;
 
 use Exception;
+use Illuminate\Container\Container;
 use Illuminate\Contracts\Queue\Job as JobContract;
 use Illuminate\Database\DetectsDeadlocks;
 use Illuminate\Queue\Jobs\Job;
@@ -22,15 +23,19 @@ class KafkaJob extends Job implements JobContract
     /**
      * KafkaJob constructor.
      *
+     * @param Container $container
      * @param KafkaQueue $connection
      * @param Message $message
-     * @param               $queue
+     * @param $connectionName
+     * @param $queue
      */
-    public function __construct(KafkaQueue $connection, Message $message, $queue)
+    public function __construct(Container $container, KafkaQueue $connection, Message $message, $connectionName, $queue)
     {
+        $this->container = $container;
         $this->connection = $connection;
-        $this->queue = $queue;
         $this->message = $message;
+        $this->connectionName = $connectionName;
+        $this->queue = $queue;
     }
 
     /**
@@ -67,7 +72,7 @@ public function fire()
      */
     public function attempts()
     {
-        return 1;
+        return (int) ($this->payload()['attempts']) + 1;
     }
 
     /**
@@ -101,7 +106,6 @@ public function release($delay = 0)
         parent::release($delay);
 
         $this->delete();
-        $this->setAttempts($this->attempts() + 1);
 
         $body = $this->payload();
 
@@ -123,16 +127,6 @@ public function release($delay = 0)
         }
     }
 
-    /**
-     * Sets the count of attempts at processing this job.
-     *
-     * @param int $count
-     */
-    private function setAttempts($count)
-    {
-        $this->connection->setAttempts($count);
-    }
-
     /**
      * Get the job identifier.
      *

src/Queue/KafkaQueue.php

@@ -31,6 +31,10 @@ class KafkaQueue extends Queue implements QueueContract
      * @var \RdKafka\KafkaConsumer
      */
     private $consumer;
+    /**
+     * @var array
+     */
+    private $subscribedQueueNames = [];
 
     /**
      * @param \RdKafka\Producer $producer
@@ -122,17 +126,24 @@ public function later($delay, $job, $data = '', $queue = null)
      */
     public function pop($queue = null)
     {
-        $this->consumer->subscribe([$this->getQueueName($queue)]);
+        $queue = $this->getQueueName($queue);
+        if (!in_array($queue, $this->subscribedQueueNames)) {
+            $this->subscribedQueueNames[] = $queue;
+            $this->consumer->subscribe($this->subscribedQueueNames);
+        }
 
-        $message = $this->consumer->consume(30 * 1000);
+        $message = $this->consumer->consume(1000);
 
         if ($message === null) {
             return;
         }
 
         switch ($message->err) {
             case RD_KAFKA_RESP_ERR_NO_ERROR:
-                return new KafkaJob($this, $message, $queue ?: $this->defaultQueue);
+                return new KafkaJob(
+                    $this->container, $this, $message,
+                    $this->connectionName, $queue ?: $this->defaultQueue
+                );
                 break;
             case RD_KAFKA_RESP_ERR__PARTITION_EOF:
             case RD_KAFKA_RESP_ERR__TIMED_OUT:

tests/KafkaQueueTest.php

@@ -7,6 +7,7 @@
 /**
  * @property \Mockery\MockInterface producer
  * @property \Mockery\MockInterface consumer
+ * @property \Mockery\MockInterface $container
  * @property array config
  * @property KafkaQueue queue
  */
@@ -18,13 +19,15 @@ public function setUp()
 
         $this->producer = Mockery::mock(\RdKafka\Producer::class);
         $this->consumer = Mockery::mock(\RdKafka\KafkaConsumer::class);
+        $this->container = Mockery::mock(\Illuminate\Container\Container::class);
 
         $this->config = [
             'queue' => str_random(),
             'sleep_error' => true,
         ];
 
         $this->queue = new KafkaQueue($this->producer, $this->consumer, $this->config);
+        $this->queue->setContainer($this->container);
     }
 
     public function test_size()

travis-install.sh

@@ -12,4 +12,4 @@ cd /tmp \
 
 pecl install rdkafka
 
-echo "extension = rdkafka.so" >> ~/.phpenv/versions/$(phpenv version-name)/etc/php.ini
+echo "extension=rdkafka.so" >> ~/.phpenv/versions/$(phpenv version-name)/etc/php.ini