Merge pull request #2 from sfcod/remove-illuminate

remove illuminate dependency, apply php 7 syntax, refactoring

Base/InteractWithTimeTrait.php

@@ -0,0 +1,75 @@
+<?php
+
+namespace SfCod\QueueBundle\Base;
+
+use DateInterval;
+use DateTime;
+use DateTimeInterface;
+
+/**
+ * Trait TimeTrait
+ *
+ * @author Virchenko Maksim <muslim1992@gmail.com>
+ *
+ * @package SfCod\QueueBundle\Job
+ */
+trait InteractWithTimeTrait
+{
+    /**
+     * Get the number of seconds until the given DateTime.
+     *
+     * @param DateTimeInterface|DateInterval|int $delay
+     *
+     * @return int
+     */
+    protected function secondsUntil($delay): int
+    {
+        $delay = $this->parseDateInterval($delay);
+
+        return $delay instanceof DateTimeInterface
+            ? max(0, $delay->getTimestamp() - $this->currentTime())
+            : (int)$delay;
+    }
+
+    /**
+     * Get the "available at" UNIX timestamp.
+     *
+     * @param DateTimeInterface|DateInterval|int $delay
+     *
+     * @return int
+     */
+    protected function availableAt($delay = 0): int
+    {
+        $delay = $this->parseDateInterval($delay);
+
+        return $delay instanceof DateTimeInterface
+            ? $delay->getTimestamp()
+            : (new DateTime())->add(new DateInterval(sprintf('PT%dS', $delay)))->getTimestamp();
+    }
+
+    /**
+     * If the given value is an interval, convert it to a DateTime instance.
+     *
+     * @param \DateTimeInterface|\DateInterval|int $delay
+     *
+     * @return \DateTimeInterface|int
+     */
+    protected function parseDateInterval($delay)
+    {
+        if ($delay instanceof DateInterval) {
+            $delay = (new DateTime())->add($delay);
+        }
+
+        return $delay;
+    }
+
+    /**
+     * Get the current system time as a UNIX timestamp.
+     *
+     * @return int
+     */
+    protected function currentTime(): int
+    {
+        return time();
+    }
+}

Base/JobInterface.php

@@ -2,7 +2,7 @@
 
 namespace SfCod\QueueBundle\Base;
 
-use Illuminate\Queue\Jobs\Job;
+use SfCod\QueueBundle\Job\JobContract;
 
 /**
  * Base interface for handlers
@@ -14,8 +14,8 @@ interface JobInterface
     /**
      * Run command from queue
      *
-     * @param Job $job
+     * @param JobContract $job
      * @param array $data
      */
-    public function fire(Job $job, array $data);
+    public function fire(JobContract $job, array $data);
 }

Base/JobQueueInterface.php

@@ -2,23 +2,13 @@
 
 namespace SfCod\QueueBundle\Base;
 
-use Illuminate\Queue\Capsule\Manager;
-use Illuminate\Queue\QueueManager;
-
 /**
  * Job queue interface
  *
  * @author Orlov Alexey <aaorlov88@gmail.com>
  */
 interface JobQueueInterface
 {
-    /**
-     * Get queue manager instance
-     *
-     * @return QueueManager
-     */
-    public function getQueueManager(): QueueManager;
-
     /**
      * Push new job to queue
      *

Base/JobResolverInterface.php

@@ -0,0 +1,20 @@
+<?php
+
+namespace SfCod\QueueBundle\Base;
+
+/**
+ * Interface JobResolverInterface
+ *
+ * @package SfCod\QueueBundle\Service
+ */
+interface JobResolverInterface
+{
+    /**
+     * Resolve job by class
+     *
+     * @param string $class
+     *
+     * @return JobInterface
+     */
+    public function resolve(string $class): JobInterface;
+}

Service/MongoDriverInterface.php → Base/MongoDriverInterface.php

@@ -1,9 +1,14 @@
 <?php
 
-namespace SfCod\QueueBundle\Service;
+namespace SfCod\QueueBundle\Base;
 
 use MongoDB\Database;
 
+/**
+ * Interface MongoDriverInterface
+ *
+ * @package SfCod\QueueBundle\Service
+ */
 interface MongoDriverInterface
 {
     /**

Command/RetryCommand.php

@@ -2,10 +2,10 @@
 
 namespace SfCod\QueueBundle\Command;
 
-use Illuminate\Queue\Jobs\Job;
-use SfCod\QueueBundle\Failer\MongoFailedJobProvider;
+use SfCod\QueueBundle\Base\MongoDriverInterface;
+use SfCod\QueueBundle\Failer\FailedJobProviderInterface;
+use SfCod\QueueBundle\Job\JobContract;
 use SfCod\QueueBundle\Service\JobQueue;
-use SfCod\QueueBundle\Service\MongoDriverInterface;
 use Symfony\Component\Console\Command\Command;
 use Symfony\Component\Console\Input\InputInterface;
 use Symfony\Component\Console\Input\InputOption;
@@ -27,7 +27,7 @@ class RetryCommand extends Command
     protected $queue;
 
     /**
-     * @var MongoFailedJobProvider
+     * @var FailedJobProviderInterface
      */
     protected $failer;
 
@@ -37,7 +37,7 @@ class RetryCommand extends Command
      * @param JobQueue $queue
      * @param MongoDriverInterface $mongoDriver
      */
-    public function __construct(JobQueue $queue, MongoFailedJobProvider $failer)
+    public function __construct(JobQueue $queue, FailedJobProviderInterface $failer)
     {
         $this->queue = $queue;
         $this->failer = $failer;
@@ -87,7 +87,7 @@ public function execute(InputInterface $input, OutputInterface $output)
     /**
      * Retry job
      *
-     * @param Job $job
+     * @param JobContract $job
      *
      * @return bool
      */

Command/RunJobCommand.php

@@ -3,8 +3,8 @@
 namespace SfCod\QueueBundle\Command;
 
 use Psr\Log\LoggerInterface;
-use SfCod\QueueBundle\Options;
-use SfCod\QueueBundle\Worker;
+use SfCod\QueueBundle\Worker\Options;
+use SfCod\QueueBundle\Worker\Worker;
 use Symfony\Component\Console\Command\Command;
 use Symfony\Component\Console\Input\InputArgument;
 use Symfony\Component\Console\Input\InputInterface;
@@ -44,7 +44,7 @@ protected function configure()
             ->setDescription('Runs single job by id.')
             ->addArgument('id', InputArgument::REQUIRED, 'The id of the job.')
             ->addOption('connection', null, InputArgument::OPTIONAL, 'The name of the connection.', 'default')
-            ->addOption('queue', null, InputArgument::OPTIONAL, 'The name of the queue.', null)
+            ->addOption('queue', null, InputArgument::OPTIONAL, 'The name of the queue.', 'default')
             ->addOption('delay', null, InputArgument::OPTIONAL, 'Delay before getting jobs.', 0)
             ->addOption('memory', null, InputArgument::OPTIONAL, 'Maximum memory usage limit.', 128)
             ->addOption('sleep', null, InputArgument::OPTIONAL, 'Sleep time before getting new job.', 3)

Command/WorkCommand.php

@@ -2,12 +2,13 @@
 
 namespace SfCod\QueueBundle\Command;
 
-use SfCod\QueueBundle\Options;
-use SfCod\QueueBundle\Worker;
+use SfCod\QueueBundle\Worker\Options;
+use SfCod\QueueBundle\Worker\Worker;
 use Symfony\Component\Console\Command\Command;
 use Symfony\Component\Console\Input\InputArgument;
 use Symfony\Component\Console\Input\InputInterface;
 use Symfony\Component\Console\Output\OutputInterface;
+use Symfony\Component\Console\Style\SymfonyStyle;
 
 /**
  * Class WorkCommand
@@ -48,7 +49,7 @@ protected function configure()
             ->addOption('maxTries', null, InputArgument::OPTIONAL, 'Max tries to run job.', 1)
             ->addOption('timeout', null, InputArgument::OPTIONAL, 'Daemon timeout.', 60)
             ->addOption('connection', null, InputArgument::OPTIONAL, 'The name of the connection.', 'default')
-            ->addOption('queue', null, InputArgument::OPTIONAL, 'The name of the queue.', null)
+            ->addOption('queue', null, InputArgument::OPTIONAL, 'The name of the queue.', 'default')
             ->setDescription('Run worker.');
     }
 
@@ -62,6 +63,8 @@ protected function configure()
      */
     public function execute(InputInterface $input, OutputInterface $output)
     {
+        $io = new SymfonyStyle($input, $output);
+
         $workerOptions = new Options(
             $input->getOption('delay'),
             $input->getOption('memory'),
@@ -72,6 +75,8 @@ public function execute(InputInterface $input, OutputInterface $output)
         $connection = $input->getOption('connection');
         $queue = $input->getOption('queue');
 
+        $io->success(sprintf('Worker daemon has started.'));
+
         $this->worker->daemon($connection, $queue, $workerOptions);
     }
 }