Merge pull request #19 from spiral/2.0-new

2.0 new

Author: wolfy-j

.travis.yml

@@ -27,6 +27,7 @@ script:
   - go test -v -race -cover ./broker/beanstalk -coverprofile=beanstalk.txt -covermode=atomic
   - go test -v -race -cover ./broker/sqs -coverprofile=sqs.txt -covermode=atomic
   - vendor/bin/phpunit --coverage-clover=coverage.xml
+  - vendor/bin/phpcs -n --standard=PSR2 --colors src tests
 
 after_success:
   - bash <(curl -s https://codecov.io/bash) -f jobs.txt

broker/beanstalk/sock.bean

@@ -15,9 +15,10 @@
     "spiral/roadrunner": "^1.1"
   },
   "require-dev": {
+    "phpunit/phpunit": "~7.0",
+    "squizlabs/php_codesniffer": "^3.4",
     "spiral/core": "^1.0",
-    "doctrine/inflector": "^1.3",
-    "phpunit/phpunit": "~7.0"
+    "doctrine/inflector": "^1.3"
   },
   "autoload": {
     "psr-4": {

composer.json

@@ -16,45 +16,44 @@
  */
 final class Consumer
 {
-    /*** @var Worker */
-    private $worker;
-
-    /** @var FactoryInterface */
-    private $factory;
+    /** @var HandlerRegistryInterface */
+    private $registry;
 
     /**
      * @codeCoverageIgnore
-     *
-     * @param Worker           $worker
-     * @param FactoryInterface $factory
+     * @param HandlerRegistryInterface $registry
      */
-    public function __construct(Worker $worker, FactoryInterface $factory)
+    public function __construct(HandlerRegistryInterface $registry)
     {
-        $this->worker = $worker;
-        $this->factory = $factory;
+        $this->registry = $registry;
     }
 
     /**
      * @codeCoverageIgnore
+     * @param Worker        $worker
      * @param callable|null $finalize
      */
-    public function serve(callable $finalize = null)
+    public function serve(Worker $worker, callable $finalize = null)
     {
-        while ($body = $this->worker->receive($context)) {
+        while ($body = $worker->receive($context)) {
             try {
                 $context = json_decode($context, true);
+                $handler = $this->registry->getHandler($context['job']);
 
-                $job = $this->factory->make($context['job'], $body);
-                $job->execute($context['id']);
+                $handler->handle(
+                    $context['job'],
+                    $context['id'],
+                    $handler->unserialize($context['job'], $body)
+                );
 
-                $this->worker->send("ok");
+                $worker->send("ok");
             } catch (\Throwable $e) {
-                $this->worker->error((string)$e);
+                $worker->error((string)$e);
             } finally {
                 if ($finalize !== null) {
                     call_user_func($finalize, $e ?? null);
                 }
             }
         }
     }
-}
+}

src/AbstractJob.php

@@ -11,4 +11,4 @@
 
 class JobException extends \RuntimeException
 {
-}
+}

src/Consumer.php

@@ -0,0 +1,43 @@
+<?php
+/**
+ * Spiral Framework.
+ *
+ * @license   MIT
+ * @author    Anton Titov (Wolfy-J)
+ */
+declare(strict_types=1);
+
+namespace Spiral\Jobs;
+
+/**
+ * Handles incoming jobs.
+ */
+interface HandlerInterface
+{
+    /**
+     * Handle incoming job.
+     *
+     * @param string $jobType
+     * @param string $jobID
+     * @param array  $payload
+     */
+    public function handle(string $jobType, string $jobID, array $payload): void;
+
+    /**
+     * Serialize payload.
+     *
+     * @param string $jobType
+     * @param array  $payload
+     * @return string
+     */
+    public function serialize(string $jobType, array $payload): string;
+
+    /**
+     * Unserialize payload.
+     *
+     * @param string $jobType
+     * @param string $payload
+     * @return array
+     */
+    public function unserialize(string $jobType, string $payload): array;
+}

src/Exception/JobException.php

@@ -0,0 +1,28 @@
+<?php
+/**
+ * Spiral Framework.
+ *
+ * @license   MIT
+ * @author    Anton Titov (Wolfy-J)
+ */
+declare(strict_types=1);
+
+namespace Spiral\Jobs;
+
+use Spiral\Jobs\Exception\JobException;
+
+/**
+ * Resolves handler for a given job type.
+ */
+interface HandlerRegistryInterface
+{
+    /**
+     * Get handler for the given job type.
+     *
+     * @param string $jobType
+     * @return HandlerInterface
+     *
+     * @throws JobException
+     */
+    public function getHandler(string $jobType): HandlerInterface;
+}

src/Factory/SpiralFactory.php

@@ -0,0 +1,65 @@
+<?php
+/**
+ * Spiral Framework.
+ *
+ * @license   MIT
+ * @author    Anton Titov (Wolfy-J)
+ */
+declare(strict_types=1);
+
+namespace Spiral\Jobs;
+
+use Spiral\Core\ResolverInterface;
+use Spiral\Jobs\Exception\JobException;
+
+/**
+ * Handler which can invoke itself.
+ */
+abstract class JobHandler implements HandlerInterface
+{
+    // default function with method injection
+    protected const HANDLE_FUNCTION = 'invoke';
+
+    /** @var ResolverInterface */
+    protected $resolver;
+
+    /**
+     * @param ResolverInterface $resolver
+     */
+    public function __construct(ResolverInterface $resolver)
+    {
+        $this->resolver = $resolver;
+    }
+
+    /**
+     * @inheritdoc
+     */
+    public function handle(string $jobType, string $jobID, array $payload): void
+    {
+        $method = new \ReflectionMethod($this, static::HANDLE_FUNCTION);
+        $method->setAccessible(true);
+
+        try {
+            $parameters = ['payload' => $payload, 'id' => $jobID,] + $payload;
+            $method->invokeArgs($this, $this->resolver->resolveArguments($method, $parameters));
+        } catch (\Throwable $e) {
+            throw new JobException(sprintf("[%s] %s", get_class($this), $e->getMessage()), $e->getCode(), $e);
+        }
+    }
+
+    /**
+     * @inheritdoc
+     */
+    public function serialize(string $jobType, array $payload): string
+    {
+        return json_encode($payload);
+    }
+
+    /**
+     * @inheritdoc
+     */
+    public function unserialize(string $jobType, string $payload): array
+    {
+        return json_decode($payload, true);
+    }
+}