Add CLI consumer interface #14

markitosgv · 2017-09-24T08:25:07Z

Add Symfony command to consume kafka messages, handling signals and providing an interface to implement your own services to consume.

Oliboy50

Thanks for this feature :)

Oliboy50 · 2017-09-24T16:25:44Z

Command/ConsumeTopicCommand.php

+    protected function configure()
+    {
+        $this
+            ->setName('kafka:consume')


maybe we should add a namespace for the command name, as it was done here => https://github.com/M6Web/PhpProcessManagerBundle/blob/master/Command/HttpProcessCommand.php#L62

so we could name it m6web:kafka:consume, what do you think?

It's ok, I'm going to add bundle namespace

Oliboy50 · 2017-09-24T16:27:11Z

Command/ConsumeTopicCommand.php

+    {
+        $this
+            ->setName('kafka:consume')
+            ->setDescription('Consume command to process kafka topic/s')


remove the / before s?

Oliboy50 · 2017-09-24T16:29:12Z

Command/ConsumeTopicCommand.php

+            ->setDescription('Consume command to process kafka topic/s')
+            ->addOption('consumer', null,InputOption::VALUE_REQUIRED, 'Consumer name')
+            ->addOption('handler', null,InputOption::VALUE_REQUIRED, 'Handler service name')
+            ->addOption('auto-commit', null,InputOption::VALUE_NONE, 'Auto commit enabled?')


add a space after null,

(we should set a php-cs-fixer check in our Travis builds 😱)

Of course, you could maybe use https://styleci.io/

Oliboy50 · 2017-09-24T16:32:38Z

Command/ConsumeTopicCommand.php

+
+    protected function execute(InputInterface $input, OutputInterface $output)
+    {
+        $prefixName = $this->getContainer()->getParameter('m6web_kafka.prefix_name');


register $container = $this->getContainer(); before this line, because this getter is used several times, and we don't know how much time consuming it is compared to a simple variable assignment

Oliboy50 · 2017-09-24T16:34:40Z

Command/ConsumeTopicCommand.php

+            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>');


(may take ...

no need to add PHP_EOL since it's already done by writeln method (except if you really want a double EOL)

source: http://api.symfony.com/3.3/Symfony/Component/Console/Output/OutputInterface.html#method_writeln

I would prefer to maintain a double EOL

Oliboy50 · 2017-09-24T16:50:47Z

Command/ConsumeTopicCommand.php

+                    $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>');


It's not an error, this appears every time you don't have any messages to consume every X seconds defined in timeout_consuming_queue

Oliboy50 · 2017-09-24T16:50:50Z

Command/ConsumeTopicCommand.php

+                    $messageHandler->process($message);
+                    break;
+                case RD_KAFKA_RESP_ERR__PARTITION_EOF:
+                    $output->writeln('<question>No more messages; will wait for more</question>');


<info>No more messages. Waiting for more...</info>?

I don't understand this, it's a message about they aren't any messages left

Oliboy50 · 2017-09-24T16:50:59Z

Command/ConsumeTopicCommand.php

+            }
+
+            if($this->shutdown) {
+                $output->writeln('<question>Shuting down...</question>');


It's about color, I don't care if you would prefer to put info :D

Oliboy50 · 2017-09-24T16:58:30Z

DependencyInjection/M6WebKafkaExtension.php

@@ -31,11 +32,14 @@ public function load(array $configs, ContainerBuilder $container)

        $this->loadProducers($container, $config);
        $this->loadConsumers($container, $config);
+
+        $container->setParameter('m6web_kafka.prefix_name', $config['prefix_services_name']);


not to sure about this prefix parameter... cc @NastasiaSaby what do you think?

anyway, if this feature is accepted:

I'd name it services_name_prefix, and you should keep the exact same name for the internal parameter name, so:

$container->setParameter('m6web_kafka.services_name_prefix', $config['services_name_prefix']);

This prefix is needed to access here:

$prefixName = $this->getContainer()->getParameter('m6web_kafka.prefix_name');

Oliboy50 · 2017-09-24T17:07:20Z

DependencyInjection/M6WebKafkaExtension.php

    }

    /**
     * @param ContainerBuilder $container
-     * @param array            $config
+     * @param array $config
+     * @throws \Symfony\Component\DependencyInjection\Exception\BadMethodCallException


IMHO this doc tag is not helpful if there's no throw statement in the method body, except if you explain in which case this exception is thrown (eg. @throws \MyException If a wrong parameter is used)

NastasiaSaby

Thank you @markitosgv for your contribution :)

NastasiaSaby · 2017-09-25T06:10:33Z

Command/ConsumeTopicCommand.php

+            }
+        }
+
+        $output->writeln('<info>End consuming topic gracefully</info>');


successfully?

NastasiaSaby · 2017-09-25T06:11:12Z

Handler/MessageHandlerInterface.php

+     */
+    public function process(Message $message);
+
+}


End of line

NastasiaSaby · 2017-09-25T06:12:20Z

README.md

+
+*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.


successfull shutdown?

NastasiaSaby · 2017-09-25T06:22:39Z

README.md

@@ -53,6 +53,7 @@ Here a configuration example:

 ```yaml
 m6_web_kafka:
+    services_name_prefix: 'custom_prefix'


Maybe, we can add a comment to say that we have the possibility to customize the service name by using our own prefix with an optional parameter.

omansour · 2017-09-25T06:46:16Z

👎

sorry @markitosgv but this will not be merged. We prefer decouple responsibilities and use the https://github.com/M6Web/DaemonBundle to build commands like this and keep this current bundle simple as possible.

omansour · 2017-09-26T07:02:16Z

after consulting the maintainers, this will not be merged. Please use the DaemonBundle as mentioned previously.

Add CLI consumer interface

02a7f8d

Oliboy50 reviewed Sep 24, 2017

View reviewed changes

Changes due to comments

5f9a4a8

NastasiaSaby reviewed Sep 25, 2017

View reviewed changes

Add memory handler and abstract implementation

aa673af

omansour closed this Sep 26, 2017

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add CLI consumer interface #14

Add CLI consumer interface #14

markitosgv commented Sep 24, 2017

Oliboy50 left a comment

Oliboy50 Sep 24, 2017

markitosgv Sep 24, 2017

Oliboy50 Sep 24, 2017

markitosgv Sep 24, 2017

Oliboy50 Sep 24, 2017

markitosgv Sep 24, 2017

Oliboy50 Sep 24, 2017

markitosgv Sep 24, 2017

Oliboy50 Sep 24, 2017

Oliboy50 Sep 24, 2017

markitosgv Sep 24, 2017

Oliboy50 Sep 24, 2017

markitosgv Sep 24, 2017

Oliboy50 Sep 24, 2017

markitosgv Sep 24, 2017

Oliboy50 Sep 24, 2017

markitosgv Sep 24, 2017

Oliboy50 Sep 24, 2017

markitosgv Sep 24, 2017

Oliboy50 Sep 24, 2017

NastasiaSaby left a comment

NastasiaSaby Sep 25, 2017

NastasiaSaby Sep 25, 2017

NastasiaSaby Sep 25, 2017

NastasiaSaby Sep 25, 2017

NastasiaSaby Sep 25, 2017

omansour commented Sep 25, 2017

omansour commented Sep 26, 2017


		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.

Add CLI consumer interface #14

Add CLI consumer interface #14

Conversation

markitosgv commented Sep 24, 2017

Oliboy50 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

NastasiaSaby left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

omansour commented Sep 25, 2017

omansour commented Sep 26, 2017