Add Command processor and Command creator

Author: jverdeyen

Diff for: pkg/enqueue/Symfony/Creator/JobCommandCreator.php

@@ -0,0 +1,66 @@
+<?php
+
+namespace Enqueue\Symfony\Creator;
+
+use Enqueue\Client\ProducerInterface;
+
+class JobCommandCreator
+{
+    /**
+     * @var ProducerInterface
+     */
+    protected $producer;
+
+    /**
+     * @var string
+     */
+    protected $env;
+
+    /**
+     * CommandJobCreator constructor.
+     *
+     * @param ProducerInterface $producer
+     * @param string            $env
+     */
+    public function __construct(ProducerInterface $producer, $env)
+    {
+        $this->producer = $producer;
+        $this->env = $env;
+    }
+
+    /**
+     * @param $command
+     * @param mixed $args
+     *
+     * @return \Enqueue\Rpc\Promise|null
+     */
+    public function scheduleCommand($command, $args = [])
+    {
+        $argumentString = $this->createArgumentString($args);
+
+        return $this->producer->sendCommand('run_command', sprintf('%s %s', $command, $argumentString));
+    }
+
+    /**
+     * @param array $arguments
+     *
+     * @return string
+     */
+    public function createArgumentString(array $arguments)
+    {
+        $optionList = [];
+
+        foreach ($arguments as $key => $value) {
+            if (!is_int($key)) {
+                $optionList[] = sprintf('--%s=%s', $key, $value);
+                continue;
+            }
+
+            $optionList[] = sprintf('%s', $value);
+        }
+
+        $optionList[] = sprintf('--env=%s', $this->env);
+
+        return implode(' ', $optionList);
+    }
+}

Diff for: pkg/enqueue/Symfony/Processor/RunCommandProcessor.php

@@ -0,0 +1,44 @@
+<?php
+
+namespace Enqueue\Symfony\Processor;
+
+use Enqueue\Client\CommandSubscriberInterface;
+use Enqueue\Consumption\Result;
+use Interop\Queue\PsrContext;
+use Interop\Queue\PsrMessage;
+use Interop\Queue\PsrProcessor;
+use Symfony\Component\Process\Exception\ProcessFailedException;
+use Symfony\Component\Process\Process;
+
+class RunCommandProcessor implements PsrProcessor, CommandSubscriberInterface
+{
+    /**
+     * @var string
+     */
+    private $projectDir;
+
+    public function __construct($projectDir)
+    {
+        $this->projectDir = $projectDir;
+    }
+
+    public function process(PsrMessage $message, PsrContext $context)
+    {
+        $commandline = $message->getBody();
+
+        $process = new Process('./bin/console '.$commandline, $this->projectDir);
+
+        try {
+            $process->mustRun();
+
+            return Result::ACK;
+        } catch (ProcessFailedException $e) {
+            return Result::reject(sprintf('The process failed with exception: "%s" in %s at %s', $e->getMessage(), $e->getFile(), $e->getLine()));
+        }
+    }
+
+    public static function getSubscribedCommand()
+    {
+        return 'run_command';
+    }
+}