Merge pull request #3 from haydenk/master

Refactor to run continuously until signal interrupt or termination

Author: mshauneu

src/Mshauneu/RdKafkaBundle/Command/TopicConsumeCommand.php

@@ -32,7 +32,7 @@ protected function configure()
 			->addOption('offset', 'o', InputOption::VALUE_OPTIONAL, 'Offset', TopicCommunicator::OFFSET_BEGINNING)
 			->addOption('timeout', 't', InputOption::VALUE_OPTIONAL, 'Timeout in ms', 1000);
 	}
-	
+
 	/**
 	 * {@inheritDoc}
 	 * @see \Symfony\Component\Console\Command\Command::execute()
@@ -68,8 +68,16 @@ protected function execute(InputInterface $input, OutputInterface $output)
         }
 
 		$topicConsumer->consumeStart($offset, $partition);
-		$topicConsumer->consume($messageHandler, $partition, $timeout);
-		$topicConsumer->consumeStop();
+
+        pcntl_signal(SIGTERM, [&$topicConsumer, 'stop']);
+        pcntl_signal(SIGINT, [&$topicConsumer, 'stop']);
+        pcntl_signal(SIGHUP, [&$topicConsumer, 'restart']);
+
+        while($topicConsumer->isConsuming()) {
+            $topicConsumer->consume($messageHandler, $partition, $timeout);
+            pcntl_signal_dispatch();
+        }
+
 	}
 
 }

src/Mshauneu/RdKafkaBundle/Topic/TopicConsumer.php

@@ -56,19 +56,39 @@ public function consume($consumer, $partition = 0, $timeoutInMs = 1000) {
 		if (true !== $this->isConsuming) {
 			throw new \Exception ("Please call consumeStart first to start consuming message");
 		}
-		
-		while ($message = $this->consumerTopic->consume($partition, $timeoutInMs)) {
-			$consumer->consume($message->topic_name, $message->partition, $message->offset, $message->key, $message->payload);
-		}
+
+        if($message = $this->consumerTopic->consume($partition, $timeoutInMs)) {
+            return $consumer->consume($message->topic_name, $message->partition, $message->offset, $message->key, $message->payload);
+        }
+
+        return true;
 	}
-	
+
+	public function stop()
+    {
+        $this->consumeStop();
+    }
+
 	/**
 	 * @param number $partition
 	 */
 	public function consumeStop($partition = 0) {
 		$this->consumerTopic->consumeStop($partition);
 		$this->consumerTopic = null;
 		$this->isConsuming = false;
-	}	
+	}
+
+	public function restart()
+    {
+        // Implement a restart method
+    }
+
+    /**
+     * @return boolean
+     */
+    public function isConsuming()
+    {
+        return $this->isConsuming;
+    }
 
 }