Add CLI consumer interface

Command/ConsumeTopicCommand.php

@@ -0,0 +1,106 @@
+<?php
+
+declare(ticks = 1);
+
+namespace M6Web\Bundle\KafkaBundle\Command;
+
+use M6Web\Bundle\KafkaBundle\Manager\ConsumerManager;
+use M6Web\Bundle\KafkaBundle\Handler\MessageHandlerInterface;
+use Symfony\Bundle\FrameworkBundle\Command\ContainerAwareCommand;
+use Symfony\Component\Console\Input\InputArgument;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Input\InputOption;
+use Symfony\Component\Console\Output\OutputInterface;
+
+class ConsumeTopicCommand extends ContainerAwareCommand
+{
+    protected $shutdown;
+
+    protected function configure()
+    {
+        $this
+            ->setName('m6web:kafka:consume')
+            ->setDescription('Consume command to process kafka topics')
+            ->addArgument('consumer', InputArgument::REQUIRED, 'Consumer name')
+            ->addArgument('handler', InputArgument::REQUIRED, 'Handler service name')
+            ->addOption('auto-commit', null, InputOption::VALUE_NONE, 'Auto commit enabled?')
+        ;
+    }
+
+    protected function execute(InputInterface $input, OutputInterface $output)
+    {
+        $container = $this->getContainer();
+        $prefixName = $container->getParameter('m6web_kafka.services_name_prefix');
+
+        $consumer = $input->getArgument('consumer');
+        $handler = $input->getArgument('handler');
+        $autoCommit = $input->getOption('auto-commit');
+
+        /**
+         * @var ConsumerManager $topicConsumer
+         */
+        $topicConsumer = $container->get(sprintf('%s.consumer.%s', $prefixName, $consumer));
+        if (!$topicConsumer) {
+            throw new \Exception(sprintf("TopicConsumer with name '%s' is not defined", $consumer));
+        }
+
+        /**
+         * @var MessageHandlerInterface $messageHandler
+         */
+        $messageHandler = $container->get($handler);
+        if (!$messageHandler) {
+            throw new \Exception(sprintf("Message Handler with name '%s' is not defined", $handler));
+        }
+
+        $output->writeln('<comment>Waiting for partition assignment... (make take some time when quickly re-joining the group after leaving it.)'.PHP_EOL.'</comment>');
+
+        $this->registerSigHandlers();
+
+        while (true) {
+            $message = $topicConsumer->consume($autoCommit);
+
+            switch ($message->err) {
+                case RD_KAFKA_RESP_ERR_NO_ERROR:
+                    $messageHandler->process($message);
+                    break;
+                case RD_KAFKA_RESP_ERR__PARTITION_EOF:
+                    $output->writeln('<question>No more messages; will wait for more</question>');
+                    break;
+                case RD_KAFKA_RESP_ERR__TIMED_OUT:
+                    $output->writeln('<question>Timed out</question>');
+                    break;
+                default:
+                    throw new \Exception($message->errstr(), $message->err);
+                    break;
+            }
+
+            if($this->shutdown) {
+                $output->writeln('<question>Shuting down...</question>');
+                if ($message->err === RD_KAFKA_RESP_ERR_NO_ERROR) {
+                    $topicConsumer->commit();
+                }
+
+                break;
+            }
+        }
+
+        $output->writeln('<info>End consuming topic gracefully</info>');
+    }
+
+    private function registerSigHandlers()
+    {
+        if(!function_exists('pcntl_signal')) {
+            return;
+        }
+
+        pcntl_signal(SIGTERM, [$this, 'shutdownFn']);
+        pcntl_signal(SIGINT, [$this, 'shutdownFn']);
+        pcntl_signal(SIGQUIT, [$this, 'shutdownFn']);
+    }
+
+    public function shutdownFn()
+    {
+        $this->shutdown = true;
+    }
+
+}

DependencyInjection/Configuration.php

@@ -20,6 +20,7 @@ public function getConfigTreeBuilder()
 
         $rootNode
             ->children()
+                ->scalarNode('services_name_prefix')->defaultValue('m6_web_kafka')->end()
                 ->booleanNode('event_dispatcher')->defaultTrue()->end()
                 ->arrayNode('consumers')
                     ->useAttributeAsKey('key')

DependencyInjection/M6WebKafkaExtension.php

@@ -20,6 +20,7 @@ class M6WebKafkaExtension extends Extension
 {
     /**
      * {@inheritDoc}
+     * @throws \Symfony\Component\DependencyInjection\Exception\BadMethodCallException
      */
     public function load(array $configs, ContainerBuilder $container)
     {
@@ -31,11 +32,14 @@ public function load(array $configs, ContainerBuilder $container)
 
         $this->loadProducers($container, $config);
         $this->loadConsumers($container, $config);
+
+        $container->setParameter('m6web_kafka.services_name_prefix', $config['services_name_prefix']);
     }
 
     /**
      * @param ContainerBuilder $container
-     * @param array            $config
+     * @param array $config
+     * @throws \Symfony\Component\DependencyInjection\Exception\BadMethodCallException
      */
     protected function loadProducers(ContainerBuilder $container, array $config)
     {
@@ -60,15 +64,15 @@ protected function loadProducers(ContainerBuilder $container, array $config)
             $this->setEventDispatcher($config, $producerDefinition);
 
             $container->setDefinition(
-                sprintf('m6_web_kafka.producer.%s', $key),
+                sprintf('%s.producer.%s', $config['services_name_prefix'], $key),
                 $producerDefinition
             );
         }
     }
 
     /**
      * @param ContainerBuilder $container
-     * @param array            $config
+     * @param array $config
      */
     protected function loadConsumers(ContainerBuilder $container, array $config)
     {
@@ -91,15 +95,16 @@ protected function loadConsumers(ContainerBuilder $container, array $config)
             $this->setEventDispatcher($config, $consumerDefinition);
 
             $container->setDefinition(
-                sprintf('m6_web_kafka.consumer.%s', $key),
+                sprintf('%s.consumer.%s', $config['services_name_prefix'], $key),
                 $consumerDefinition
             );
         }
     }
 
     /**
-     * @param array      $config
+     * @param array $config
      * @param Definition $definition
+     * @throws \Symfony\Component\DependencyInjection\Exception\InvalidArgumentException
      */
     protected function setEventDispatcher(array $config, Definition $definition)
     {

Handler/MessageHandlerInterface.php

@@ -0,0 +1,17 @@
+<?php
+
+namespace M6Web\Bundle\KafkaBundle\Handler;
+
+use RdKafka\Message;
+
+interface MessageHandlerInterface
+{
+    /**
+     * Process message from kafka
+     *
+     * @param Message $message
+     * @return mixed
+     */
+    public function process(Message $message);
+
+}

README.md

@@ -53,6 +53,7 @@ Here a configuration example:
 
 ```yaml
 m6_web_kafka:
+    services_name_prefix: 'custom_prefix'
     event_dispatcher: true
     producers:
        producer1:
@@ -165,6 +166,8 @@ But you can choose not to do it by adding an argument as following:
 $consumer->consume(false);
 ```
 
+*Note:* Setting $consumer->consume(false) boost consuming performance, avoiding to commit every single kafka consumed message.
+
 You can decide to commit manually your message with:
 ```php
 $consumer->commit();
@@ -178,6 +181,43 @@ It is the `\RdKafka\Message` from the [RdKafka extension](https://arnaud-lb.gith
 In case there is no more message, it will give you a _No more message_ string.
 In case there is a time out, it will give you a _Time out_ string.
 
+#### Consuming from CLI
+
+You can use your own service to process Kakfa topic messages and run it from Symfony special command provided by us.
+
+At first you have to create your own service implementing MessageHandlerInterface. For example:
+
+```php
+<?php
+
+namespace AppBundle\Service;
+
+use M6Web\Bundle\KafkaBundle\Handler\MessageHandlerInterface;
+use RdKafka\Message;
+
+class TestConsumer implements MessageHandlerInterface
+{
+
+    public function process(Message $message)
+    {
+        echo $message->payload. PHP_EOL;
+    }
+
+}
+```
+
+Then you could start consuming messages executing this command with your consumer and service name:
+
+```
+php bin/console kafka:consume you_consumer_name "AppBundle\Service\TestConsumer"
+```
+
+Also you can add --auto-commit option to enable auto commit in every consumed message.
+
+*Note:* If you aren't using *Symfony Automatic Service Loading*, introduced in Symfony 3.3, you have to put in handler option your service name instead of FQCN.
+
+This Symfony command provide *signal management* too, handling SIGQUIT, SIGTERM and SIGINT to do a gracefully shutdown, committing last processed message and exiting.
+
 ### Exceptions list
 - EntityNotSetException
 - KafkaException