feat: support native consumer with cli consumer (golang) (#79)

* feat: support native consumer with cli consumer (golang)

* fix: correct yml & register command

* fix: use the correct service

* fix: correct negative logic

Author: gsdevme

Command/RabbitMqConsumerCommand.php

@@ -0,0 +1,67 @@
+<?php
+
+namespace Markup\JobQueueBundle\Command;
+
+use OldSound\RabbitMqBundle\RabbitMq\ConsumerInterface;
+use PhpAmqpLib\Message\AMQPMessage;
+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;
+
+/**
+ * Consumes messages from the Go consumer
+ */
+class RabbitMqConsumerCommand extends ContainerAwareCommand
+{
+    const STRICT_CODE_ACK = 0;
+    const STRICT_CODE_REJECT = 3;
+    const STRICT_CODE_REJECT_REQUEUE = 4;
+    const STRICT_CODE_NEG_ACK = 5;
+    const STRICT_CODE_NEG_ACK_REQUEUE = 6;
+
+    protected function configure()
+    {
+        $this
+            ->addArgument('event', InputArgument::REQUIRED)
+            ->setName('markup:job_queue:rabbitmq_consumer')
+            ->addOption(
+                'strict-exit-code',
+                null,
+                InputOption::VALUE_NONE,
+                'If strict_exit_code is chosen then this command will return the following exit codes. 0=ACK, 3=REJECT, 4=REJECT & REQUEUE, 5=NEG ACK, 6=NEG ACK & REQUEUE'
+            );
+    }
+
+    /**
+     * {inheritDoc}
+     */
+    protected function execute(InputInterface $input, OutputInterface $output)
+    {
+        $data = json_decode(base64_decode($input->getArgument('event')), true);
+
+        $strict = $input->getOption('strict-exit-code');
+
+        /** @var ConsumerInterface $consumer */
+        $consumer = $this->getContainer()->get('simple_bus.rabbit_mq_bundle_bridge.commands_consumer');
+
+        if (!$consumer instanceof ConsumerInterface) {
+            return 0;
+        }
+
+        $consumerReturn = $consumer->execute(new AMQPMessage($data['body'], $data['properties']));
+
+        // if not running in strict mode - always acknowledge the message otherwise it will requeue forever
+        if (!$strict) {
+            exit(self::STRICT_CODE_ACK);
+        }
+
+        // If in strict mode then test the return value from the consumer and return an appropriate code
+        if ($consumerReturn === ConsumerInterface::MSG_REJECT) {
+            exit(self::STRICT_CODE_REJECT);
+        }
+
+        exit(self::STRICT_CODE_ACK);
+    }
+}

DependencyInjection/Configuration.php

@@ -30,8 +30,11 @@ public function getConfigTreeBuilder()
                          ->addDefaultsIfNotSet()
                          ->children()
                              ->integerNode('prefetch_count')
-                                 ->defaultValue(1)
-                                 ->min(1)
+                                 ->defaultValue(5)
+                                 ->min(5)
+                             ->end()
+                             ->scalarNode('consumer')
+                                 ->defaultValue('markup:job_queue:consumer')
                              ->end()
                          ->end()
                      ->end()

Resources/config/commands.yml

@@ -38,4 +38,7 @@ services:
     markup_job_queue.command.write_supervisord_config_file:
         class: Markup\JobQueueBundle\Command\WriteSupervisordConfigFileCommand
         tags:
-            - { name: console.command, command: 'markup:job_queue:supervisord_config:write'' }
+            - { name: console.command, command: 'markup:job_queue:supervisord_config:write' }
+    Markup\JobQueueBundle\Command\RabbitMqConsumerCommand:
+        tags:
+            - { name: console.command, command: 'markup:job_queue:rabbitmq_consumer' }

Service/SupervisordConfigFileWriter.php

@@ -7,17 +7,6 @@
 
 class SupervisordConfigFileWriter
 {
-
-    /**
-     * Symfony command that receives message from RabbitMQ and processes it (cli consumer)
-     */
-    const CLI_CONSUMPTION_COMMAND = 'markup:job_queue:consumer';
-
-    /**
-     * Symfony command that receives message from RabbitMQ and processes it (php consumer)
-     */
-    const PHP_CONSUMPTION_COMMAND = 'rabbitmq:consumer';
-
     const MODE_PHP = 'php';
     const MODE_CLI = 'cli';
 
@@ -161,7 +150,7 @@ public function getConfigForCliConsumer($uniqueEnvironment, $skipExistsChecks =
                 '%s -e "%s/console %s --strict-exit-code --env=%s --no-debug" -c %s -V -i --strict-exit-code',
                 $this->consumerPath,
                 $kernelPath,
-                self::CLI_CONSUMPTION_COMMAND,
+                $topicConfig['consumer'],
                 $this->kernelEnv,
                 $cliConfigFile
             );
@@ -201,7 +190,7 @@ public function getConfigForPhpConsumer($uniqueEnvironment, $skipExistsChecks =
             $consumerCommand = sprintf(
                 '%s/console %s -m %s %s --env=%s --no-debug',
                 $this->kernelPath,
-                self::PHP_CONSUMPTION_COMMAND,
+                'rabbitmq:consumer',
                 $topicConfig['prefetch_count'],
                 $topic,
                 $this->kernelEnv

Tests/Service/SupervisordConfigFileWriterTest.php

@@ -18,8 +18,8 @@ protected function setUp()
             '/etc/rabbitmq-cli-consumer/config'
         );
 
-        $topicA = ['prefetch_count' => 1];
-        $topicB = ['prefetch_count' => 2];
+        $topicA = ['prefetch_count' => 1, 'consumer' => 'markup:job_queue:consumer'];
+        $topicB = ['prefetch_count' => 2, 'consumer' => 'markup:job_queue:consumer'];
         $this->writer->setTopicsConfiguration(['testqueuea' => $topicA, 'testqueueb' => $topicB]);
     }