Ag

README.md

@@ -5,29 +5,31 @@
 Задача:
 
 * Шедулер (/src/Scheduler) должен предоставлять:
-    * интерфейс для добавления заданий в расписание;
-    * интерфейс для получения заданий из расписания для выполнения;
+    + интерфейс для добавления заданий в расписание;
+    + интерфейс для получения заданий из расписания для выполнения;
     * интерфейсы для добавления и получения заданий из очереди;
     * интерфейс для фабрики обработчиков заданий;
     * интерфейс для обработчика задания;
     * базовую реализацию воркера для выборки заданий из расписания и добавления их в очередь;
     * базовую реализацию воркера для получения задания из очереди, поиска обработчика для неё и запуска этого обработчика.
-* Шедулер должен запускаться по крону раз в минуту, выбирать из расписания задания, которые нужно выполнить и добавлять их очередь на выполнения.
+
+* Шедулер должен запускаться по крону раз в минуту, выбирать из расписания задания, которые нужно выполнить и добавлять их очередь на выполнение.
 * Базовые реализации интерфейсов должны быть покрыты тестами.
+
 * Основной код (/src/Topface) должен предоставлять:
-    * возможность запуска крона через run.php;
-    * возможность запуска обработчиков очереди через run.php;
-    * возможность добавлять два-три типа задания в расписание;
-    * реализации необходимых для работы библиотеки интерфейсов;
-    * обработчики заданий расписания.
+    + возможность запуска крона через run.php;
+    + возможность запуска обработчиков очереди через run.php;
+    + возможность добавлять два-три типа задания в расписание;
+    + реализации необходимых для работы библиотеки интерфейсов;
+    + обработчики заданий расписания.
 * Способ просмотра результатов работы заданий можно выбрать самим (логи, ES и т.п.). 
-* Для хранения расписания, заданий и очереди использовать Redis.
++ Для хранения расписания, заданий и очереди использовать Redis.
 * Опционально:
-    * Добавить логирование действия через PSR-3 интерфейс.
+    + Добавить логирование действия через PSR-3 интерфейс.
     * Добавить сбор статистических метрик работы шедулера.
 
 Для упрощения работы добавлены:
 * [интерфейс](/src/Scheduler/Task/SchedulerTaskInterface.php) и [реализация](/src/Scheduler/Task/SchedulerTask.php) для сущности "Задание";
 * [интерфейс](/src/Scheduler/SchedulerInterface.php) и [простой вариант реализации](/src/Scheduler/Scheduler.php) логики хранения расписания.
 
-Для запуска кода и тестов советуется использовать докер-образ hub.core.tf/topface-ci:1.4.
+Для запуска кода и тестов советуется использовать докер-образ hub.core.tf/topface-ci:1.4.

composer.json

@@ -16,7 +16,8 @@
     "php": "^7.0",
     "ext-msgpack": "~2.0.0",
     "predis/predis": "v1.0.1",
-    "php-di/php-di": "~6.0.0"
+    "php-di/php-di": "~6.0.0",
+    "monolog/monolog": "~1.23"
   },
   "require-dev": {
     "phpunit/phpunit": "~6.5"

src/Scheduler/Handler/HandlerFactoryInterface.php

@@ -0,0 +1,15 @@
+<?php
+
+namespace Scheduler\Handler;
+
+/**
+ * Фабрика обработчиков очереди
+ */
+interface HandlerFactoryInterface {
+    /**
+     * @param int $typeId Тип таска для обработки
+     *
+     * @return HandlerInterface
+     */
+    public function getHandler(int $typeId): HandlerInterface;
+}

src/Scheduler/Handler/HandlerInterface.php

@@ -0,0 +1,19 @@
+<?php
+
+namespace Scheduler\Handler;
+
+use Scheduler\Task\SchedulerTask;
+
+/**
+ * Интерфейс обработчика таски
+ */
+interface HandlerInterface {
+    /**
+     * Обрабатываем таску и возвращаем результат выполнения
+     *
+     * @param SchedulerTask $SchedulerTask
+     *
+     * @return bool
+     */
+    public function runTask(SchedulerTask $SchedulerTask): bool;
+}

src/Scheduler/HandlerWorker/HandlerWorker.php

@@ -0,0 +1,108 @@
+<?php
+
+namespace Scheduler\HandlerWorker;
+
+use Exception;
+use Psr\Log\LoggerInterface;
+use Scheduler\Handler\HandlerFactoryInterface;
+use Scheduler\SchedulerQueueStorage\SchedulerQueueStorageInterface;
+use Scheduler\Task\SchedulerTask;
+
+/**
+ * Обрабатываем очередь задач, которые нам нужно выполнить
+ */
+class HandlerWorker implements HandlerWorkerInterface {
+    /**
+     * Сторейдж доступа к очереди тасков на выполнение
+     *
+     * @var SchedulerQueueStorageInterface
+     */
+    private $SchedulerQueueStorage;
+
+    /**
+     * Фабрика обработчиков типов тасков
+     *
+     * @var HandlerFactoryInterface
+     */
+    private $HandlerFactory;
+
+    /**
+     * Логгер для записи событий
+     *
+     * @var LoggerInterface
+     */
+    private $Logger;
+
+    /**
+     * @param SchedulerQueueStorageInterface $SchedulerQueueStorage
+     * @param HandlerFactoryInterface        $HandlerFactory
+     * @param LoggerInterface                $Logger
+     */
+    public function __construct(
+        SchedulerQueueStorageInterface $SchedulerQueueStorage,
+        HandlerFactoryInterface $HandlerFactory,
+        LoggerInterface $Logger
+    ) {
+        $this->SchedulerQueueStorage = $SchedulerQueueStorage;
+        $this->HandlerFactory = $HandlerFactory;
+        $this->Logger = $Logger;
+    }
+
+    /**
+     * Получаем новое задание из очереди и обрабатываем его
+     */
+    public function run() {
+        while(true) {
+            // достаем элемент из очереди
+            $Task = $this->SchedulerQueueStorage->pop();
+
+            if (!$Task) {
+                continue;
+            }
+
+            $result = $this->processTask($Task);
+
+            if (!$result) {
+                $this->Logger->error(
+                    sprintf(
+                        'Task #%s failed and restarted',
+                        $Task->getTaskId()
+                    )
+                );
+                $this->SchedulerQueueStorage->add($Task);
+            }
+
+            // чистим память в обработчике, чтобы не текла
+            unset($Task);
+
+            // чуть-чуть спим для приличия
+            usleep(500);
+        }
+    }
+
+    /**
+     * Получаем хэндлер для задачи и выполняем её
+     *
+     * @param SchedulerTask $SchedulerTask Задача, которую необходимо выполнить
+     *
+     * @return bool
+     */
+    public function processTask(SchedulerTask $SchedulerTask): bool {
+        $TaskHandler = $this->HandlerFactory->getHandler($SchedulerTask->getTypeId());
+        try {
+            return $TaskHandler->runTask($SchedulerTask);
+        } catch (Exception $Ex) {
+            $this->Logger->error(
+                sprintf(
+                    'Task #%s handler raise an exception %s, message: %s\n%s',
+                    $SchedulerTask->getTaskId(),
+                    get_class($Ex),
+                    $Ex->getMessage(),
+                    $Ex->getTraceAsString()
+                )
+            );
+        }
+
+        return false;
+    }
+}

src/Scheduler/HandlerWorker/HandlerWorkerInterface.php

@@ -0,0 +1,13 @@
+<?php
+
+namespace Scheduler\HandlerWorker;
+
+/**
+ * Интерфейс обработчика очереди
+ */
+interface HandlerWorkerInterface {
+    /**
+     * Получаем новое задание из очереди и обрабатываем его
+     */
+    public function run();
+}

src/Scheduler/Scheduler.php

@@ -6,8 +6,10 @@
 use Scheduler\Task\SchedulerTask;
 use Scheduler\Task\SchedulerTaskInterface;
 
+/**
+ * Класс расписания выполняемых задач
+ */
 class Scheduler implements SchedulerInterface {
-
     /**
      * @var Client
      */
@@ -18,6 +20,8 @@ public function __construct(SchedulerRedisClientFactoryInterface $RedisClientFac
     }
 
     /**
+     * Добавляем задачу в пул или устанавливаем новые данные для неё
+     *
      * @inheritdoc
      */
     public function addOrSet(SchedulerTaskInterface $Task): string {
@@ -30,6 +34,8 @@ public function addOrSet(SchedulerTaskInterface $Task): string {
     }
 
     /**
+     * Достаем таск из очереди и удаляем его оттуда
+     *
      * @inheritdoc
      */
     public function getAndRemove(int $timestamp, int $count = self::LIMIT): array {