FuzeWorks/Async
FuzeWorks
/
Async
1
0
Fork 0
Code Issues 5 Pull Requests Releases 1 Activity

Finished Async into a working state (#1) 

Initial release.

Removed Tracy dependency

Implemented dependencyConstraint.

Partially fixes #3.

Built foundation.

System is now fully in a working order, though many parts must still be developed.
System still requires the following components:
- Dependency Constraint. Only run when all dependencies have ran.
- TaskScheduler. Standardized task that plans task periodically and maintains consistency.
- Better TaskStorage.
- Proper documentation

Brought everything back into working condition.

Also updated all copyright notices.

Moved files into right place.

Co-authored-by: Abel Hoogeveen <abel@techfuze.net>
Reviewed-on: #1
This commit is contained in:
Abel Hoogeveen 2020-02-01 14:32:36 +01:00
parent 8ccb179643
commit d35f15114b
27 changed files with 1287 additions and 559 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
.gitignore vendored Normal file
View File

@ -0,0 +1,5 @@
composer.phar
composer.lock
.idea/
log/
vendor/

7
Dockerfile Normal file
View File

@ -0,0 +1,7 @@
FROM php:7.3-cli-buster
RUN apt-get update &&\
apt-get install --no-install-recommends --assume-yes --quiet procps ca-certificates curl git &&\
rm -rf /var/lib/apt/lists/*
# PDO
RUN docker-php-ext-install pdo_mysql

211
Tasks/src/Supervisors/ParallelSuperVisor.php
View File

@ -1,211 +0,0 @@
<?php
/**
* FuzeWorks CLIComponent.
*
* The FuzeWorks PHP FrameWork
*
* Copyright (C) 2013-2019 TechFuze
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
* SOFTWARE.
*
* @author TechFuze
* @copyright Copyright (c) 2013 - 2019, TechFuze. (http://techfuze.net)
* @license https://opensource.org/licenses/MIT MIT License
*
* @link http://techfuze.net/fuzeworks
* @since Version 1.2.0
*
* @version Version 1.2.0
*/
namespace Application\Library\Tasks\Supervisors;
use Application\Library\Tasks\Executor;
use Application\Library\Tasks\SuperVisor;
use Application\Library\Tasks\Task;
use Application\Library\Tasks\TasksException;
use Application\Library\Tasks\TaskStorage;
class ParallelSuperVisor implements SuperVisor
{
/**
* @var TaskStorage
*/
protected $taskStorage;
/**
* @var Executor
*/
protected $executor;
/**
* @var Task[]
*/
protected $tasks = [];
public function __construct(TaskStorage $taskStorage, Executor $executor, array $options = [])
{
$this->taskStorage = $taskStorage;
$this->executor = $executor;
}
/**
* @inheritDoc
*/
public function cycle(): int
{
// First: if there are no tasks, load them
$this->taskStorage->refreshTasks();
$this->tasks = $this->taskStorage->readTasks();
// If there are still no tasks, nothing is queued, so this cycle can end.
if (empty($this->tasks))
return SuperVisor::FINISHED;
// Check if all tasks are completed
$allCompleted = true;
foreach ($this->tasks as $task)
if ($task->getStatus() !== Task::COMPLETED)
$allCompleted = false;
if ($allCompleted)
return SuperVisor::FINISHED;
for ($i=0;$i<count($this->tasks);$i++)
{
$task = $this->tasks[$i];
dump(date('H:i:s') . ': ' . $task->getId() .'/'.$i . ": state #" . $task->getStatus());
// DELAYED Tasks. If a task is delayed, but the time has passed, mark the task as pending
if ($task->getStatus() === Task::DELAYED && time() < $task->getDelayTime())
$task->setStatus(Task::PENDING);
// FAILED Tasks, check if failed can be made pending using constraints
elseif ($task->getStatus() === Task::FAILED)
$task = $this->testConstraints($task);
// FINISHED/CANCELLED/FAILED Tasks, check if they are old enough for removal
elseif ($task->getStatus() === Task::COMPLETED || $task->getStatus() === Task::CANCELLED || $task->getStatus() === Task::FAILED)
{
}
// If the task is pending, start if not constrained
elseif ($task->getStatus() === Task::PENDING)
{
// First test for constraints.
$task = $this->testConstraints($task);
// If the task is still PENDING, execute it
if ($task->getStatus() !== Task::PENDING)
continue;
// START THE PROCESS
$task = $this->executor->startTask($task);
$this->taskStorage->modifyTask($task);
}
// Check if tasks are still running
elseif ($task->getStatus() === Task::RUNNING)
{
$isRunning = $this->executor->getTaskRunning($task);
$output = $this->taskStorage->readTaskOutput($task);
$hasOutput = !is_null($output);
if (!$isRunning && !$hasOutput)
$task->setStatus(Task::FAILED);
elseif (!$isRunning && $hasOutput)
{
try {
$task->setOutput($output['output'], $output['errors']);
$task->setStatus($output['statusCode']);
} catch (TasksException $e) {
// Ignore
}
}
// elseif (EXPIRED)
else
continue;
$this->taskStorage->modifyTask($task);
}
// Check if task succeeded, and whether the task must progress to POST or COMPLETED
// @todo Add failed handler
elseif ($task->getStatus() === Task::SUCCESS)
{
// Calling post handler when requested to do so
if ($task->getCallPostHandler() === true || ($task->getCallPostHandlerWhenFailed() === true && $output['statusCode'] === Task::FAILED))
$task = $this->executor->startTask($task, true);
else
$task->setStatus(Task::COMPLETED);
$this->taskStorage->modifyTask($task);
}
// If the task is in post mode, check when that process has ended
elseif ($task->getStatus() === Task::POST)
{
$isRunning = $this->executor->getTaskRunning($task);
$output = $this->taskStorage->readTaskOutput($task);
$hasOutput = isset($output['postStatus']);
if (!$isRunning && !$hasOutput)
$task->setStatus(Task::FAILED);
elseif (!$isRunning && $hasOutput)
{
try {
$task->setPostOutput($output['postOutput'], $output['postErrors']);
$task->setStatus(Task::COMPLETED);
} catch (TasksException $e) {
// Ignore
}
}
else
continue;
$this->taskStorage->modifyTask($task);
}
}
return SuperVisor::RUNNING;
}
private function testConstraints(Task $task): Task
{
$constraints = $task->getConstraints();
foreach ($constraints as $constraint)
{
if ($constraint->intervene($task) && $constraint->blockCode() != 0)
{
$task->setStatus($constraint->blockCode());
if ($constraint->blockCode() === Task::DELAYED)
$task->setDelayTime(time() + $constraint->delayTime());
// Save changes to TaskStorage
$this->taskStorage->modifyTask($task);
}
}
return $task;
}
}

84
Tasks/supervisor.php
View File

@ -1,84 +0,0 @@
<?php
/**
* FuzeWorks Application Skeleton.
*
* The FuzeWorks PHP FrameWork
*
* Copyright (C) 2013-2019 TechFuze
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
* SOFTWARE.
*
* @author TechFuze
* @copyright Copyright (c) 2013 - 2019, TechFuze. (http://techfuze.net)
* @license https://opensource.org/licenses/MIT MIT License
*
* @link http://techfuze.net/fuzeworks
* @since Version 0.0.1
*
* @version Version 1.2.0
*/
use Application\Library\Tasks;
use Application\Library\Tasks\Executors\ShellExecutor;
use Application\Library\Tasks\SuperVisor;
use Application\Library\Tasks\Supervisors\ParallelSuperVisor;
use Application\Library\Tasks\TaskStorage\ArrayTaskStorage;
use FuzeWorks\Logger;
use FuzeWorks\Priority;
use FuzeWorks\WebAppComponent;
require_once(dirname(__DIR__) . DIRECTORY_SEPARATOR . 'vendor/autoload.php');
// Open configurator
$configurator = new FuzeWorks\Configurator();
// Set up basic settings
$configurator->setTimeZone('Europe/Amsterdam');
$configurator->setTempDirectory(dirname(__DIR__) . DIRECTORY_SEPARATOR . 'temp');
$configurator->setLogDirectory(dirname(__DIR__). DIRECTORY_SEPARATOR . 'log');
// Enable components
// WebComponent
$webAppComponent = new WebAppComponent();
$configurator->addComponent($webAppComponent);
// Add directories
$configurator->addDirectory(dirname(__FILE__) . DIRECTORY_SEPARATOR . 'Config', 'config', Priority::HIGH);
$configurator->addDirectory(dirname(__FILE__) . DIRECTORY_SEPARATOR . 'Helper', 'helpers', Priority::HIGH);
$configurator->addDirectory(dirname(__FILE__) . DIRECTORY_SEPARATOR . 'Library', 'libraries', Priority::HIGH);
$configurator->addDirectory(dirname(__FILE__) . DIRECTORY_SEPARATOR . 'Plugin', 'plugins', Priority::HIGH);
$configurator->addDirectory(dirname(__FILE__) . DIRECTORY_SEPARATOR . 'Controller', 'controllers', Priority::HIGH);
$configurator->addDirectory(dirname(__FILE__) . DIRECTORY_SEPARATOR . 'View', 'views', Priority::HIGH);
$configurator->addDirectory(dirname(__FILE__) . DIRECTORY_SEPARATOR . 'Model', 'models', Priority::HIGH);
$configurator->addDirectory(dirname(__FILE__) . DIRECTORY_SEPARATOR . 'Layout', 'layouts', Priority::NORMAL);
// Debug
$configurator->enableDebugMode()->setDebugAddress('ALL');
$container = $configurator->createContainer();
Logger::enableScreenLog();
// RUN THE APP
/** @var Tasks $lib */
$lib = $container->libraries->get('tasks');
$supervisor = $lib->getSuperVisor();
while ($supervisor->cycle() === SuperVisor::RUNNING) {
usleep(250000);
}

99
bin/supervisor Normal file
View File

@ -0,0 +1,99 @@
#!/usr/bin/env php
<?php
/**
* FuzeWorks Async Library
*
* The FuzeWorks PHP FrameWork
*
* Copyright (C) 2013-2020 TechFuze
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
* SOFTWARE.
*
* @author TechFuze
* @copyright Copyright (c) 2013 - 2020, TechFuze. (http://techfuze.net)
* @license https://opensource.org/licenses/MIT MIT License
*
* @link http://techfuze.net/fuzeworks
* @since Version 1.0.0
*
* @version Version 1.0.0
*/
use FuzeWorks\Async\SuperVisor;
use FuzeWorks\Async\Tasks;
use FuzeWorks\Async\TasksException;
use FuzeWorks\Exception\InvalidArgumentException;
use FuzeWorks\Exception\LibraryException;
if (version_compare('7.1.0', PHP_VERSION, '>')) {
fwrite(
STDERR,
sprintf(
'FuzeWorks Async requires PHP 7.1 or higher.' . PHP_EOL .
'You are using PHP %s (%s).' . PHP_EOL,
PHP_VERSION,
PHP_BINARY
)
);
die(1);
}
// First load composer
$autoloaders = [
__DIR__ . '/../../autoload.php',
__DIR__ . '/../vendor/autoload.php',
__DIR__ . '/vendor/autoload.php'
];
foreach ($autoloaders as $file)
if (file_exists($file))
require($file);
try {
// Open configurator
$configurator = new FuzeWorks\Configurator();
// Set up basic settings
$configurator->setTimeZone('Europe/Amsterdam');
$configurator->setTempDirectory(dirname(__FILE__) . DIRECTORY_SEPARATOR . 'temp');
$configurator->setLogDirectory(dirname(__FILE__). DIRECTORY_SEPARATOR . 'log');
// Add Async library
$configurator->deferComponentClassMethod('libraries', 'addLibraryClass', null, 'async', '\FuzeWorks\Async\Tasks');
// Debug
$configurator->enableDebugMode()->setDebugAddress('ALL');
// Create container
$container = $configurator->createContainer();
// RUN THE APP
/** @var Tasks $lib */
$lib = $container->libraries->get('async');
$supervisor = $lib->getSuperVisor();
while ($supervisor->cycle() === SuperVisor::RUNNING) {
usleep(250000);
}
} catch (InvalidArgumentException | TasksException | LibraryException $e) {
fwrite(STDERR, sprintf('FuzeWorks Async could not load.' . PHP_EOL .
'Exception: ' . $e->getMessage() . PHP_EOL)
);
}

26
composer.json Normal file
View File

@ -0,0 +1,26 @@
{
"name": "fuzeworks/async",
"description": "FuzeWorks Async task execution library",
"license": ["MIT"],
"authors": [
{
"name": "TechFuze",
"homepage": "https://techfuze.net"
},
{
"name": "FuzeWorks Community",
"homepage": "https://techfuze.net/fuzeworks/contributors"
}
],
"require": {
"php": ">=7.2.0",
"fuzeworks/core": "~1.2.0",
"ext-json": "*"
},
"autoload": {
"psr-4": {
"FuzeWorks\\Async\\": "src/FuzeWorks/Async"
}
},
"bin": ["bin/supervisor"]
}

12
Tasks/config.tasks.php → config.tasks.php
View File

@ -1,10 +1,10 @@
<?php <?php
/** /**
* FuzeWorks CLIComponent. * FuzeWorks Async Library
* *
* The FuzeWorks PHP FrameWork * The FuzeWorks PHP FrameWork
* *
* Copyright (C) 2013-2019 TechFuze * Copyright (C) 2013-2020 TechFuze
* *
* Permission is hereby granted, free of charge, to any person obtaining a copy * Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal * of this software and associated documentation files (the "Software"), to deal
@ -25,13 +25,13 @@
* SOFTWARE. * SOFTWARE.
* *
* @author TechFuze * @author TechFuze
* @copyright Copyright (c) 2013 - 2019, TechFuze. (http://techfuze.net) * @copyright Copyright (c) 2013 - 2020, TechFuze. (http://techfuze.net)
* @license https://opensource.org/licenses/MIT MIT License * @license https://opensource.org/licenses/MIT MIT License
* *
* @link http://techfuze.net/fuzeworks * @link http://techfuze.net/fuzeworks
* @since Version 1.2.0 * @since Version 1.0.0
* *
* @version Version 1.2.0 * @version Version 1.0.0
*/ */
return array( return array(
'SuperVisor' => [ 'SuperVisor' => [
@ -48,6 +48,6 @@ return array(
'type' => 'ShellExecutor', 'type' => 'ShellExecutor',
// For ShellExecutor, first parameter is the file location of the worker script // For ShellExecutor, first parameter is the file location of the worker script
'parameters' => [dirname(dirname(__DIR__)) . DS . 'worker.php'] 'parameters' => [dirname(__FILE__) . DS . 'worker.php']
] ]
); );

15
Tasks/src/Constraint.php → src/FuzeWorks/Async/Constraint.php
View File

@ -1,10 +1,10 @@
<?php <?php
/** /**
* FuzeWorks CLIComponent. * FuzeWorks Async Library
* *
* The FuzeWorks PHP FrameWork * The FuzeWorks PHP FrameWork
* *
* Copyright (C) 2013-2019 TechFuze * Copyright (C) 2013-2020 TechFuze
* *
* Permission is hereby granted, free of charge, to any person obtaining a copy * Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal * of this software and associated documentation files (the "Software"), to deal
@ -25,17 +25,16 @@
* SOFTWARE. * SOFTWARE.
* *
* @author TechFuze * @author TechFuze
* @copyright Copyright (c) 2013 - 2019, TechFuze. (http://techfuze.net) * @copyright Copyright (c) 2013 - 2020, TechFuze. (http://techfuze.net)
* @license https://opensource.org/licenses/MIT MIT License * @license https://opensource.org/licenses/MIT MIT License
* *
* @link http://techfuze.net/fuzeworks * @link http://techfuze.net/fuzeworks
* @since Version 1.2.0 * @since Version 1.0.0
* *
* @version Version 1.2.0 * @version Version 1.0.0
*/ */
namespace Application\Library\Tasks; namespace FuzeWorks\Async;
interface Constraint interface Constraint
{ {
@ -61,6 +60,8 @@ interface Constraint
/** /**
* In case blockCode() returns Task::DELAYED, this method shall return the actual delay time * In case blockCode() returns Task::DELAYED, this method shall return the actual delay time
* *
* Must provide a unix timestamp when to retry execution.
*
* @return int * @return int
*/ */
public function delayTime(): int; public function delayTime(): int;

169
src/FuzeWorks/Async/Constraint/DependencyConstraint.php Normal file
View File

@ -0,0 +1,169 @@
<?php
/**
* FuzeWorks Async Library
*
* The FuzeWorks PHP FrameWork
*
* Copyright (C) 2013-2020 TechFuze
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
* SOFTWARE.
*
* @author TechFuze
* @copyright Copyright (c) 2013 - 2020, TechFuze. (http://techfuze.net)
* @license https://opensource.org/licenses/MIT MIT License
*
* @link http://techfuze.net/fuzeworks
* @since Version 1.0.0
*
* @version Version 1.0.0
*/
namespace FuzeWorks\Async\Constraint;
use FuzeWorks\Async\Constraint;
use FuzeWorks\Async\Task;
use FuzeWorks\Async\Tasks;
use FuzeWorks\Async\TasksException;
use FuzeWorks\Exception\FactoryException;
use FuzeWorks\Exception\LibraryException;
use FuzeWorks\Factory;
use FuzeWorks\Libraries;
use FuzeWorks\Logger;
/**
* Class DependencyConstraint
*
* Use this constraint to prevent a task from running if one or multiple tasks this task depends on hasn't completed yet.
* This constraint also can be applied to queue multiple tasks. Make a chain of tasks that depend on one another.
*
* @package FuzeWorks\Async\Constraint
*/
class DependencyConstraint implements Constraint
{
public $dependencies = [];
protected $delayTimes = 3;
protected $returnStatus = Task::DELAYED;
/**
* DependencyConstraint constructor.
*
* To add dependencies, the following array should be supplied:
* $dependencyList: array(string 'taskId', string 'taskId', string 'taskId')
*
* @param array $dependencyList
* @param int $delayTimes Time that a task should be delayed before retrying
* @throws TasksException
*/
public function __construct(array $dependencyList, int $delayTimes = 3)
{
$this->dependencies = $dependencyList;
$this->delayTimes = $delayTimes;
// Test if async library can be loaded
$this->loadTasksLib();
}
/**
* @inheritDoc
*/
public function intervene(Task $task): bool
{
// Fetch taskStorage
try {
$tasks = $this->loadTasksLib();
$taskStorage = $tasks->getTaskStorage();
// Is any dependency unresolved?
$hasUnresolved = false;
// Test if any dependency has not been resolved
foreach ($this->dependencies as $dependency)
{
// Get dependency
$dependencyTask = $taskStorage->getTaskById($dependency);
// If the dependency task is completed, ignore it and continue to next dependency
if ($dependencyTask->getStatus() === Task::COMPLETED)
continue;
elseif ($dependencyTask->getStatus() === Task::CANCELLED)
{
// Cancel current task
$task->setOutput('', 'Task cancelled due to failed dependency.');
$this->returnStatus = Task::CANCELLED;
return true;
}
else
$hasUnresolved = true;
}
// If any unresolved tasks exist, delay task execution
if ($hasUnresolved)
{
$this->returnStatus = Task::DELAYED;
return true;
}
} catch (TasksException $e) {
Logger::logError("Could not constraint task. '" . $e->getMessage() . "'");
$this->returnStatus = Task::FAILED;
return true;
}
return false;
}
/**
* @inheritDoc
*/
public function blockCode(): int
{
return $this->returnStatus;
}
/**
* @inheritDoc
*/
public function delayTime(): int
{
return time() + $this->delayTimes;
}
/**
* Load the tasks library, so that dependencies can get scanned later
*
* @return Tasks
* @throws TasksException
*/
private function loadTasksLib(): Tasks
{
try {
/** @var Libraries $libraries */
$libraries = Factory::getInstance('libraries');
/** @var Tasks $tasks */
$tasks = $libraries->get('async');
return $tasks;
} catch (FactoryException | LibraryException $e) {
throw new TasksException("Could not constrain task. Async library could not be loaded.");
}
}
}

96
src/FuzeWorks/Async/Constraint/FixedTimeConstraint.php Normal file
View File

@ -0,0 +1,96 @@
<?php
/**
* FuzeWorks Async Library
*
* The FuzeWorks PHP FrameWork
*
* Copyright (C) 2013-2020 TechFuze
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
* SOFTWARE.
*
* @author TechFuze
* @copyright Copyright (c) 2013 - 2020, TechFuze. (http://techfuze.net)
* @license https://opensource.org/licenses/MIT MIT License
*
* @link http://techfuze.net/fuzeworks
* @since Version 1.0.0
*
* @version Version 1.0.0
*/
namespace FuzeWorks\Async\Constraint;
use FuzeWorks\Async\Constraint;
use FuzeWorks\Async\Task;
/**
* Class FixedTimeConstraint
*
* Use this constraint to prevent the execution of a task before a certain time.
* This constraint is particularly useful for planning and spreading the execution of tasks.
*
* @package FuzeWorks\Async\Constraint
*/
class FixedTimeConstraint implements Constraint
{
/**
* @var int
*/
public $timestamp;
/**
* FixedTimeConstraint constructor.
*
* Provide a unix timestamp for when this task should be run.
* The SuperVisor will start the task if the time has exceeded the provided timestamp
*
* @param int $unitTimeStamp Unix time to start this task
*/
public function __construct(int $unitTimeStamp)
{
$this->timestamp = $unitTimeStamp;
}
/**
* @inheritDoc
*/
public function intervene(Task $task): bool
{
if ($task->getStatus() === Task::PENDING && time() < $this->timestamp)
return true;
return false;
}
/**
* @inheritDoc
*/
public function blockCode(): int
{
return Task::DELAYED;
}
/**
* @inheritDoc
*/
public function delayTime(): int
{
return $this->timestamp;
}
}

14
Tasks/src/Events/TaskHandleEvent.php → src/FuzeWorks/Async/Events/TaskHandleEvent.php
View File

@ -1,10 +1,10 @@
<?php <?php
/** /**
* FuzeWorks CLIComponent. * FuzeWorks Async Library
* *
* The FuzeWorks PHP FrameWork * The FuzeWorks PHP FrameWork
* *
* Copyright (C) 2013-2019 TechFuze * Copyright (C) 2013-2020 TechFuze
* *
* Permission is hereby granted, free of charge, to any person obtaining a copy * Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal * of this software and associated documentation files (the "Software"), to deal
@ -25,17 +25,17 @@
* SOFTWARE. * SOFTWARE.
* *
* @author TechFuze * @author TechFuze
* @copyright Copyright (c) 2013 - 2019, TechFuze. (http://techfuze.net) * @copyright Copyright (c) 2013 - 2020, TechFuze. (http://techfuze.net)
* @license https://opensource.org/licenses/MIT MIT License * @license https://opensource.org/licenses/MIT MIT License
* *
* @link http://techfuze.net/fuzeworks * @link http://techfuze.net/fuzeworks
* @since Version 1.2.0 * @since Version 1.0.0
* *
* @version Version 1.2.0 * @version Version 1.0.0
*/ */
namespace Application\Library\Tasks\Events; namespace FuzeWorks\Async\Events;
use Application\Library\Tasks\Task; use FuzeWorks\Async\Task;
use FuzeWorks\Event; use FuzeWorks\Event;
class TaskHandleEvent extends Event class TaskHandleEvent extends Event

13
Tasks/src/Executor.php → src/FuzeWorks/Async/Executor.php
View File

@ -1,10 +1,10 @@
<?php <?php
/** /**
* FuzeWorks CLIComponent. * FuzeWorks Async Library
* *
* The FuzeWorks PHP FrameWork * The FuzeWorks PHP FrameWork
* *
* Copyright (C) 2013-2019 TechFuze * Copyright (C) 2013-2020 TechFuze
* *
* Permission is hereby granted, free of charge, to any person obtaining a copy * Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal * of this software and associated documentation files (the "Software"), to deal
@ -25,17 +25,16 @@
* SOFTWARE. * SOFTWARE.
* *
* @author TechFuze * @author TechFuze
* @copyright Copyright (c) 2013 - 2019, TechFuze. (http://techfuze.net) * @copyright Copyright (c) 2013 - 2020, TechFuze. (http://techfuze.net)
* @license https://opensource.org/licenses/MIT MIT License * @license https://opensource.org/licenses/MIT MIT License
* *
* @link http://techfuze.net/fuzeworks * @link http://techfuze.net/fuzeworks
* @since Version 1.2.0 * @since Version 1.0.0
* *
* @version Version 1.2.0 * @version Version 1.0.0
*/ */
namespace Application\Library\Tasks; namespace FuzeWorks\Async;
interface Executor interface Executor
{ {

86
Tasks/src/Executors/ShellExecutor.php → src/FuzeWorks/Async/Executors/ShellExecutor.php
View File

@ -1,10 +1,10 @@
<?php <?php
/** /**
* FuzeWorks CLIComponent. * FuzeWorks Async Library
* *
* The FuzeWorks PHP FrameWork * The FuzeWorks PHP FrameWork
* *
* Copyright (C) 2013-2019 TechFuze * Copyright (C) 2013-2020 TechFuze
* *
* Permission is hereby granted, free of charge, to any person obtaining a copy * Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal * of this software and associated documentation files (the "Software"), to deal
@ -25,19 +25,20 @@
* SOFTWARE. * SOFTWARE.
* *
* @author TechFuze * @author TechFuze
* @copyright Copyright (c) 2013 - 2019, TechFuze. (http://techfuze.net) * @copyright Copyright (c) 2013 - 2020, TechFuze. (http://techfuze.net)
* @license https://opensource.org/licenses/MIT MIT License * @license https://opensource.org/licenses/MIT MIT License
* *
* @link http://techfuze.net/fuzeworks * @link http://techfuze.net/fuzeworks
* @since Version 1.2.0 * @since Version 1.0.0
* *
* @version Version 1.2.0 * @version Version 1.0.0
*/ */
namespace Application\Library\Tasks\Executors; namespace FuzeWorks\Async\Executors;
use Application\Library\Tasks\Executor; use FuzeWorks\Async\Executor;
use Application\Library\Tasks\Task; use FuzeWorks\Async\Process;
use Application\Library\Tasks\TasksException; use FuzeWorks\Async\Task;
use FuzeWorks\Async\TasksException;
class ShellExecutor implements Executor class ShellExecutor implements Executor
{ {
@ -64,7 +65,6 @@ class ShellExecutor implements Executor
$this->worker = $workerFile; $this->worker = $workerFile;
} }
private function shellExec($format, array $parameters = []) private function shellExec($format, array $parameters = [])
{ {
$parameters = array_map("escapeshellarg", $parameters); $parameters = array_map("escapeshellarg", $parameters);
@ -81,18 +81,8 @@ class ShellExecutor implements Executor
// Then execute the command using the base64_encoded string of the taskID // Then execute the command using the base64_encoded string of the taskID
$output = $this->shellExec($commandString, [base64_encode($task->getId())]); $output = $this->shellExec($commandString, [base64_encode($task->getId())]);
$pid = intval($output[0]);
// Add the PID to the task and change its state to 'RUNNING' $task->setProcess(new Process($pid));
if ($post)
{
$task->setStatus(Task::POST);
$task->addAttribute('post_pid', $output[0]);
}
else
{
$task->setStatus(Task::RUNNING);
$task->addAttribute('run_pid', $output[0]);
}
// And finally return the task // And finally return the task
return $task; return $task;
@ -114,28 +104,38 @@ class ShellExecutor implements Executor
// First prepare the command used to gather info on processes // First prepare the command used to gather info on processes
$commandString = "ps -o pid,%%cpu,%%mem,state,start -p %s | sed 1d"; $commandString = "ps -o pid,%%cpu,%%mem,state,start -p %s | sed 1d";
// Then fetch the PID // First we must determine what process is used.
try { $process = $task->getProcess();
$pid = is_null($task->attribute('post_pid')) ? $task->attribute('run_pid') : $task->attribute('post_pid'); if (is_null($process))
$output = $this->shellExec($commandString, [$pid]);
if (count($output) < 1)
return null;
$last = $output[count($output) - 1];
if (trim($last) === "")
return null;
$parts = preg_split("/\s+/", trim($last));
$pid = intval($parts[0]);
$state = strtoupper(trim($parts[3]));
if ("{$pid}" !== $parts[0] || $state === 'Z')
return null;
return $parts;
} catch (TasksException $e) {
return null; return null;
}
// Then using that process we determine the ProcessID
$pid = $process->getPid();
// And we execute the commandString to fetch info on the process
$output = $this->shellExec($commandString, [$pid]);
// If not output is provided, the command failed and should return null
if (count($output) < 1)
return null;
// ??
$last = $output[count($output) - 1];
if (trim($last) === "")
return null;
// Split up the info
$parts = preg_split("/\s+/", trim($last));
// Determine the state of the process
// If the process is in a 'zombie' state, it should be considered fully executed.
// Cleanup of Zombie processes must take place by periodically restarting the SuperVisor, or by using a SuperVisor which does not have the zombie problem
$state = strtoupper(trim($parts[3]));
if ("{$pid}" !== $parts[0] || $state === 'Z')
return null;
// Finally, return the Task information
return $parts;
} }
public function getTaskExitCode(Task $task): int public function getTaskExitCode(Task $task): int

15
Tasks/src/Handler.php → src/FuzeWorks/Async/Handler.php
View File

@ -1,10 +1,10 @@
<?php <?php
/** /**
* FuzeWorks CLIComponent. * FuzeWorks Async Library
* *
* The FuzeWorks PHP FrameWork * The FuzeWorks PHP FrameWork
* *
* Copyright (C) 2013-2019 TechFuze * Copyright (C) 2013-2020 TechFuze
* *
* Permission is hereby granted, free of charge, to any person obtaining a copy * Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal * of this software and associated documentation files (the "Software"), to deal
@ -25,16 +25,16 @@
* SOFTWARE. * SOFTWARE.
* *
* @author TechFuze * @author TechFuze
* @copyright Copyright (c) 2013 - 2019, TechFuze. (http://techfuze.net) * @copyright Copyright (c) 2013 - 2020, TechFuze. (http://techfuze.net)
* @license https://opensource.org/licenses/MIT MIT License * @license https://opensource.org/licenses/MIT MIT License
* *
* @link http://techfuze.net/fuzeworks * @link http://techfuze.net/fuzeworks
* @since Version 1.2.0 * @since Version 1.0.0
* *
* @version Version 1.2.0 * @version Version 1.0.0
*/ */
namespace Application\Library\Tasks; namespace FuzeWorks\Async;
interface Handler interface Handler
{ {
@ -63,9 +63,8 @@ interface Handler
* *
* Remember: postHandler is called in a new process. Data is not shared between processes unless done manually. * Remember: postHandler is called in a new process. Data is not shared between processes unless done manually.
* @param Task $task * @param Task $task
* @return bool
*/ */
public function postHandler(Task $task): bool; public function postHandler(Task $task);
/** /**
* Any output generated by postHandler should be returned here * Any output generated by postHandler should be returned here

73
src/FuzeWorks/Async/Process.php Normal file
View File

@ -0,0 +1,73 @@
<?php
/**
* FuzeWorks Async Library
*
* The FuzeWorks PHP FrameWork
*
* Copyright (C) 2013-2020 TechFuze
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
* SOFTWARE.
*
* @author TechFuze
* @copyright Copyright (c) 2013 - 2020, TechFuze. (http://techfuze.net)
* @license https://opensource.org/licenses/MIT MIT License
*
* @link http://techfuze.net/fuzeworks
* @since Version 1.0.0
*
* @version Version 1.0.0
*/
namespace FuzeWorks\Async;
class Process
{
const PENDING = 1;
const RUNNING = 2;
const FAILED = 3;
const FINISHED = 4;
/**
* The current status of the process
*
* @var int
*/
protected $status = Process::PENDING;
/**
* @var int
*/
protected $pid;
public function __construct(int $pid)
{
$this->pid = $pid;
}
/**
* Receive the process Id of this process
*
* @return int
*/
public function getPid(): int
{
return $this->pid;
}
}

12
Tasks/src/SuperVisor.php → src/FuzeWorks/Async/SuperVisor.php
View File

@ -1,10 +1,10 @@
<?php <?php
/** /**
* FuzeWorks CLIComponent. * FuzeWorks Async Library
* *
* The FuzeWorks PHP FrameWork * The FuzeWorks PHP FrameWork
* *
* Copyright (C) 2013-2019 TechFuze * Copyright (C) 2013-2020 TechFuze
* *
* Permission is hereby granted, free of charge, to any person obtaining a copy * Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal * of this software and associated documentation files (the "Software"), to deal
@ -25,16 +25,16 @@
* SOFTWARE. * SOFTWARE.
* *
* @author TechFuze * @author TechFuze
* @copyright Copyright (c) 2013 - 2019, TechFuze. (http://techfuze.net) * @copyright Copyright (c) 2013 - 2020, TechFuze. (http://techfuze.net)
* @license https://opensource.org/licenses/MIT MIT License * @license https://opensource.org/licenses/MIT MIT License
* *
* @link http://techfuze.net/fuzeworks * @link http://techfuze.net/fuzeworks
* @since Version 1.2.0 * @since Version 1.0.0
* *
* @version Version 1.2.0 * @version Version 1.0.0
*/ */
namespace Application\Library\Tasks; namespace FuzeWorks\Async;
interface SuperVisor interface SuperVisor

273
src/FuzeWorks/Async/Supervisors/ParallelSuperVisor.php Normal file
View File

@ -0,0 +1,273 @@
<?php
/**
* FuzeWorks Async Library
*
* The FuzeWorks PHP FrameWork
*
* Copyright (C) 2013-2020 TechFuze
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
* SOFTWARE.
*
* @author TechFuze
* @copyright Copyright (c) 2013 - 2020, TechFuze. (http://techfuze.net)
* @license https://opensource.org/licenses/MIT MIT License
*
* @link http://techfuze.net/fuzeworks
* @since Version 1.0.0
*
* @version Version 1.0.0
*/
namespace FuzeWorks\Async\Supervisors;
use FuzeWorks\Async\Executor;
use FuzeWorks\Async\SuperVisor;
use FuzeWorks\Async\Task;
use FuzeWorks\Async\TasksException;
use FuzeWorks\Async\TaskStorage;
class ParallelSuperVisor implements SuperVisor
{
/**
* @var TaskStorage
*/
protected $taskStorage;
/**
* @var Executor
*/
protected $executor;
/**
* @var Task[]
*/
protected $tasks = [];
public function __construct(TaskStorage $taskStorage, Executor $executor, array $options = [])
{
$this->taskStorage = $taskStorage;
$this->executor = $executor;
}
/**
* @inheritDoc
*/
public function cycle(): int
{
// First: if there are no tasks, load them
$this->taskStorage->refreshTasks();
$this->tasks = $this->taskStorage->readTasks();
// If there are still no tasks, nothing is queued, so this cycle can end.
if (empty($this->tasks))
return SuperVisor::FINISHED;
for ($i=0;$i<count($this->tasks);$i++)
{
$task = $this->tasks[$i];
// PENDING: should start if not constrained
if ($task->getStatus() === Task::PENDING)
{
// Test if constrained
$task = $this->testConstraints($task);
// If the task changed status, task is no longer pending and should be processed by another statement
if ($task->getStatus() !== Task::PENDING)
continue;
// Start the process using the executor service
$task = $this->executor->startTask($task);
$task->setStatus(Task::RUNNING);
// Modify the task in TaskStorage
$this->taskStorage->modifyTask($task);
}
// DELAYED: If task is delayed, and enough time has passed, change the status back to pending
elseif ($task->getStatus() === Task::DELAYED && time() > $task->getDelayTime())
{
$task->setStatus(Task::PENDING);
$this->taskStorage->modifyTask($task);
}
// CANCELLED/COMPLETED: remove the task if requested to do so
elseif ($task->getStatus() === Task::COMPLETED || $task->getStatus() === Task::CANCELLED)
{
}
// RUNNING: check if task is still running. If not, set result based on output
elseif ($task->getStatus() === Task::RUNNING)
{
$isRunning = $this->executor->getTaskRunning($task);
$output = $this->taskStorage->readTaskOutput($task, $task->getRetries());
$hasOutput = !is_null($output);
// If nothing is found, the process has crashed and status PFAILED should be set
if (!$isRunning && !$hasOutput)
$task->setStatus(Task::PFAILED);
// @todo Set PFAILED after $max_Time
// If output is found, use the status code from that
elseif (!$isRunning && $hasOutput)
{
try {
$task->setOutput($output['output'], $output['errors']);
$task->setStatus($output['statusCode']);
} catch (TasksException $e) {
// On failure to set output, consider as a process failure
$task->setStatus(Task::PFAILED);
}
}
// In any other situation the process is still running and should be left alone
else
continue;
// If any changes have been made, they should be written to TaskStorage
$this->taskStorage->modifyTask($task);
}
// FAILED: if a process has failed, attempt to rety if requested to do so
elseif ($task->getStatus() === Task::PFAILED || $task->getStatus() === Task::FAILED)
{
// First fetch retry conditions
$settings = $task->getRetrySettings();
// First test if any retries should be tried at all
if ($settings['retryOnFail'] === true && $task->getRetries() < $settings['maxRetries'])
{
// Then test if this type of failure should be retried and whether the mexRetries has been exceeded
if (
($task->getStatus() === Task::PFAILED && $settings['retryPFailures'] === true) ||
($task->getStatus() === Task::FAILED && $settings['retryRFailures'] === true)
)
{
// If eligible, reset task to pending
$task->addRetry();
$task = $this->executor->startTask($task);
$task->setStatus(Task::RUNNING);
$this->taskStorage->modifyTask($task);
continue;
}
}
// If the task is not eligible for a retry, either cancel it or move it to a postHandler
if ($task->getUsePostHandler() === true)
{
$task->resetRetries();
$task = $this->executor->startTask($task, true);
$task->setStatus(Task::POST);
}
else
$task->setStatus(Task::CANCELLED);
$this->taskStorage->modifyTask($task);
}
// SUCCESS: if a task has succeeded, see if it needs a postHandler
elseif ($task->getStatus() === Task::SUCCESS)
{
if ($task->getUsePostHandler() === true)
{
$task->resetRetries();
$task = $this->executor->startTask($task, true);
$task->setStatus(Task::POST);
}
else
$task->setStatus(Task::COMPLETED);
$this->taskStorage->modifyTask($task);
}
// POST: when a task is currently running in it's postHandler
elseif ($task->getStatus() === Task::POST)
{
$isRunning = $this->executor->getTaskRunning($task);
$output = $this->taskStorage->readPostOutput($task, $task->getRetries());
$hasOutput = !is_null($output);
// If a task is not running and has no output, an error has occurred
if (!$isRunning && !$hasOutput)
{
// Test if a retry should be attempted
$settings = $task->getRetrySettings();
if ($settings['retryOnFail'] === true && $settings['retryPostFailures'] === true && $settings['maxRetries'] > $task->getRetries())
{
$task->addRetry();
$task = $this->executor->startTask($task, true);
}
elseif ($settings['maxRetries'] <= $task->getRetries())
$task->setStatus(Task::CANCELLED);
}
// @todo Retry after $max_Time
// If a task is not running and has output, set that output and mark as completed
elseif (!$isRunning && $hasOutput)
{
$task->setPostOutput($output['output'], $output['errors']);
$task->setStatus(Task::COMPLETED);
}
// If the task is still running, leave it be
else
continue;
// If any changes have been made, they should be written to TaskStorage
$this->taskStorage->modifyTask($task);
}
}
// Check if all tasks are completed
$allCompleted = true;
$anyDelayed = false;
foreach ($this->tasks as $task)
{
if ($task->getStatus() !== Task::COMPLETED && $task->getStatus() !== Task::CANCELLED)
$allCompleted = false;
elseif ($task->getStatus() === Task::DELAYED)
$anyDelayed = true;
}
// If all are finished and none are delayed
if ($allCompleted && !$anyDelayed)
return SuperVisor::FINISHED;
if ($allCompleted && $anyDelayed)
return SuperVisor::CONSTRAINED;
else
return SuperVisor::RUNNING;
}
private function testConstraints(Task $task): Task
{
$constraints = $task->getConstraints();
foreach ($constraints as $constraint)
{
if ($constraint->intervene($task) && $constraint->blockCode() != 0)
{
$task->setStatus($constraint->blockCode());
if ($constraint->blockCode() === Task::DELAYED)
$task->setDelayTime($constraint->delayTime());
// Save changes to TaskStorage
$this->taskStorage->modifyTask($task);
}
}
return $task;
}
}

268
Tasks/src/Task.php → src/FuzeWorks/Async/Task.php
View File

@ -1,10 +1,10 @@
<?php <?php
/** /**
* FuzeWorks CLIComponent. * FuzeWorks Async Library
* *
* The FuzeWorks PHP FrameWork * The FuzeWorks PHP FrameWork
* *
* Copyright (C) 2013-2019 TechFuze * Copyright (C) 2013-2020 TechFuze
* *
* Permission is hereby granted, free of charge, to any person obtaining a copy * Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal * of this software and associated documentation files (the "Software"), to deal
@ -25,27 +25,90 @@
* SOFTWARE. * SOFTWARE.
* *
* @author TechFuze * @author TechFuze
* @copyright Copyright (c) 2013 - 2019, TechFuze. (http://techfuze.net) * @copyright Copyright (c) 2013 - 2020, TechFuze. (http://techfuze.net)
* @license https://opensource.org/licenses/MIT MIT License * @license https://opensource.org/licenses/MIT MIT License
* *
* @link http://techfuze.net/fuzeworks * @link http://techfuze.net/fuzeworks
* @since Version 1.2.0 * @since Version 1.0.0
* *
* @version Version 1.2.0 * @version Version 1.0.0
*/ */
namespace Application\Library\Tasks; namespace FuzeWorks\Async;
class Task class Task
{ {
/**
* If a task is waiting to be executed, its status will be PENDING
*/
const PENDING = 1; const PENDING = 1;
/**
* If a task is currently in process, its status will be RUNNING
*/
const RUNNING = 2; const RUNNING = 2;
/**
* If a task does not reach the goal it sets out to, its status should be set to FAILED.
*
* Not to be confused with Process::FAILED. Process::FAILED will be the Process status when a Process running a Task unexpectedly stops running.
* This will set the Task to Task::PFAILED
*/
const FAILED = 3; const FAILED = 3;
const SUCCESS = 4; const PFAILED = 4;
const POST = 5;
const COMPLETED = 6; /**
const DELAYED = 7; * If a task has reached its goal, its status should be set to SUCCESS
const CANCELLED = 8; */
const SUCCESS = 5;
/**
* If a task is in process of its postHandler, its status will be POST
*/
const POST = 6;
/**
* If a task has completed all its goals, its status will be COMPLETED
*/
const COMPLETED = 7;
/**
* If a task has a timed based constraint, its status will be DELAYED
*
* Task is delayed by the time in $delayTime
*/
const DELAYED = 8;
/**
* If a task has a constraint preventing it from running at any point, its status will be CANCELLED
*/
const CANCELLED = 9;
public static function getStatusType($statusId)
{
switch ($statusId) {
case 1:
return 'Task::PENDING';
case 2:
return 'Task::RUNNING';
case 3:
return 'Task::FAILED';
case 4:
return 'Task::PFAILED';
case 5:
return 'Task::SUCCESS';
case 6:
return 'Task::POST';
case 7:
return 'Task::COMPLETED';
case 8:
return 'Task::DELAYED';
case 9:
return 'Task::CANCELLED';
default:
return false;
}
}
/** /**
* @var string * @var string
@ -57,6 +120,11 @@ class Task
*/ */
protected $handlerClass; protected $handlerClass;
/**
* @var bool
*/
protected $usePostHandler = false;
/** /**
* @var array * @var array
*/ */
@ -100,21 +168,21 @@ class Task
* *
* @var array * @var array
*/ */
protected $data = []; protected $attributes = [];
/** /**
* Whether the post handler should be called after every task execution * @var Process
*
* @var bool
*/ */
protected $callPostHandler = false; protected $process;
/** /* -------- Some settings ------------ */
* Whether the post handler should be called after execution has failed
* protected $retryOnFail = false;
* @var bool protected $maxRetries = 2;
*/ protected $retryPFailures = true;
protected $callPostHandlerWhenFailed = false; protected $retryRFailures = true;
protected $retryPostFailures = true;
protected $retries = 0;
/** /**
* Task constructor. * Task constructor.
@ -123,13 +191,15 @@ class Task
* *
* @param string $identifier The unique identifier of this task. Make sure it is always unique! * @param string $identifier The unique identifier of this task. Make sure it is always unique!
* @param string $handlerClass The class that shall handle this task * @param string $handlerClass The class that shall handle this task
* @param bool $usePostHandler Whether the postHandler on handlerClass should also be used
* @param mixed $parameters,... The arguments provided to the method that shall handle this class * @param mixed $parameters,... The arguments provided to the method that shall handle this class
* @throws TasksException * @throws TasksException
*/ */
public function __construct(string $identifier, string $handlerClass) public function __construct(string $identifier, string $handlerClass, bool $usePostHandler = false)
{ {
$this->taskId = $identifier; $this->taskId = $identifier;
$this->handlerClass = $handlerClass; $this->handlerClass = $handlerClass;
$this->usePostHandler = $usePostHandler;
if (func_num_args() > 3) if (func_num_args() > 3)
$args = array_slice(func_get_args(), 3); $args = array_slice(func_get_args(), 3);
else else
@ -162,6 +232,16 @@ class Task
return $this->handlerClass; return $this->handlerClass;
} }
/**
* Whether the postHandler on the handlerClass should be invoked after processing the initial task.
*
* @return bool
*/
public function getUsePostHandler(): bool
{
return $this->usePostHandler;
}
/** /**
* Gets the arguments to be provided to the method of the class that shall process this task * Gets the arguments to be provided to the method of the class that shall process this task
* *
@ -205,6 +285,8 @@ class Task
/** /**
* Sets the status of this Task. * Sets the status of this Task.
* *
* Must be one of the constants of this Task class
*
* @param int $status * @param int $status
*/ */
public function setStatus(int $status) public function setStatus(int $status)
@ -240,10 +322,10 @@ class Task
*/ */
public function attribute(string $key) public function attribute(string $key)
{ {
if (!isset($this->data[$key])) if (!isset($this->attributes[$key]))
return null; return null;
return $this->data[$key]; return $this->attributes[$key];
} }
/** /**
@ -258,7 +340,7 @@ class Task
if (!$this->isSerializable($value)) if (!$this->isSerializable($value))
throw new TasksException("Could not set Task '$this->taskId' attribute '$key'. Value not serializable."); throw new TasksException("Could not set Task '$this->taskId' attribute '$key'. Value not serializable.");
$this->data[$key] = $value; $this->attributes[$key] = $value;
} }
/** /**
@ -276,31 +358,6 @@ class Task
return $this->postOutput; return $this->postOutput;
} }
/**
* Sets the conditions for when the Handler::postHandler should be called.
*
* If you want the postHandler to only get called on errors, set onEvery to false and onFail to true.
*
* @param bool $onEvery Call the post handler on every result
* @param bool $onFail Call the post handler only when failed
*/
public function setPostHandler(bool $onEvery = false, bool $onFail = false)
{
$this->callPostHandler = $onEvery;
$this->callPostHandlerWhenFailed = $onFail;
}
public function getCallPostHandler(): bool
{
return $this->callPostHandler;
}
public function getCallPostHandlerWhenFailed(): bool
{
return $this->callPostHandlerWhenFailed;
}
/** /**
* Return the errors of this task execution * Return the errors of this task execution
* *
@ -317,24 +374,121 @@ class Task
} }
/**
* @todo Handle output from multiple attempts
* @param string $output
* @param string $errors
*/
public function setOutput(string $output, string $errors) public function setOutput(string $output, string $errors)
{ {
if (!is_null($this->output) || !is_null($this->errors))
throw new TasksException("Could not set output. Output already set.");
$this->output = $output; $this->output = $output;
$this->errors = $errors; $this->errors = $errors;
} }
/**
* @todo Handle output from multiple attempts
* @param string $output
* @param string $errors
*/
public function setPostOutput(string $output, string $errors) public function setPostOutput(string $output, string $errors)
{ {
if (!is_null($this->postOutput) || !is_null($this->postErrors))
throw new TasksException("Could not set post output. Output already set.");
$this->postOutput = $output; $this->postOutput = $output;
$this->postErrors = $errors; $this->postErrors = $errors;
} }
/**
* Sets the initial process for this task.
*
* To be set by Executor
*
* @param Process $process
*/
public function setProcess(Process $process)
{
$this->process = $process;
}
/**
* Returns the initial process for this task
*
* @return Process|null
*/
public function getProcess(): ?Process
{
return $this->process;
}
public function removeProcess(): bool
{
if ($this->process instanceof Process)
{
$this->process = null;
return true;
}
return false;
}
/**
* Set whether this task should retry after a failure, and how many times
*
* @param bool $retryOnFail
* @param int $maxRetries
* @param bool $retryRegularFailures
* @param bool $retryProcessFailures
* @param bool $retryPostFailures
*/
public function setRetrySettings(bool $retryOnFail, int $maxRetries = 2, bool $retryRegularFailures = true, bool $retryProcessFailures = true, bool $retryPostFailures = true)
{
$this->retryOnFail = $retryOnFail;
$this->maxRetries = $maxRetries;
$this->retryPFailures = $retryProcessFailures;
$this->retryRFailures = $retryRegularFailures;
$this->retryPostFailures = $retryPostFailures;
}
/**
* Returns the failure retry settings
*
* @return array
*/
public function getRetrySettings(): array
{
return [
'retryOnFail' => $this->retryOnFail,
'maxRetries' => $this->maxRetries,
'retryPFailures' => $this->retryPFailures,
'retryRFailures' => $this->retryRFailures,
'retryPostFailures' => $this->retryPostFailures
];
}
/**
* Add a retry to the retry counter
*/
public function addRetry()
{
$this->retries++;
}
/**
* Reset the retry counter back to 0
*/
public function resetRetries()
{
$this->retries = 0;
}
/**
* Receive the amount of retries already attempted
*
* @return int
*/
public function getRetries(): int
{
return $this->retries;
}
/** /**
* Checks whether an object can be serialized * Checks whether an object can be serialized
* *

12
Tasks/src/TaskFailException.php → src/FuzeWorks/Async/TaskFailException.php
View File

@ -1,10 +1,10 @@
<?php <?php
/** /**
* FuzeWorks CLIComponent. * FuzeWorks Async Library
* *
* The FuzeWorks PHP FrameWork * The FuzeWorks PHP FrameWork
* *
* Copyright (C) 2013-2019 TechFuze * Copyright (C) 2013-2020 TechFuze
* *
* Permission is hereby granted, free of charge, to any person obtaining a copy * Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal * of this software and associated documentation files (the "Software"), to deal
@ -25,16 +25,16 @@
* SOFTWARE. * SOFTWARE.
* *
* @author TechFuze * @author TechFuze
* @copyright Copyright (c) 2013 - 2019, TechFuze. (http://techfuze.net) * @copyright Copyright (c) 2013 - 2020, TechFuze. (http://techfuze.net)
* @license https://opensource.org/licenses/MIT MIT License * @license https://opensource.org/licenses/MIT MIT License
* *
* @link http://techfuze.net/fuzeworks * @link http://techfuze.net/fuzeworks
* @since Version 1.2.0 * @since Version 1.0.0
* *
* @version Version 1.2.0 * @version Version 1.0.0
*/ */
namespace Application\Library\Tasks; namespace FuzeWorks\Async;
class TaskFailException extends TasksException class TaskFailException extends TasksException
{ {

64
Tasks/src/TaskStorage.php → src/FuzeWorks/Async/TaskStorage.php
View File

@ -1,10 +1,10 @@
<?php <?php
/** /**
* FuzeWorks CLIComponent. * FuzeWorks Async Library
* *
* The FuzeWorks PHP FrameWork * The FuzeWorks PHP FrameWork
* *
* Copyright (C) 2013-2019 TechFuze * Copyright (C) 2013-2020 TechFuze
* *
* Permission is hereby granted, free of charge, to any person obtaining a copy * Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal * of this software and associated documentation files (the "Software"), to deal
@ -25,16 +25,16 @@
* SOFTWARE. * SOFTWARE.
* *
* @author TechFuze * @author TechFuze
* @copyright Copyright (c) 2013 - 2019, TechFuze. (http://techfuze.net) * @copyright Copyright (c) 2013 - 2020, TechFuze. (http://techfuze.net)
* @license https://opensource.org/licenses/MIT MIT License * @license https://opensource.org/licenses/MIT MIT License
* *
* @link http://techfuze.net/fuzeworks * @link http://techfuze.net/fuzeworks
* @since Version 1.2.0 * @since Version 1.0.0
* *
* @version Version 1.2.0 * @version Version 1.0.0
*/ */
namespace Application\Library\Tasks; namespace FuzeWorks\Async;
interface TaskStorage interface TaskStorage
{ {
@ -88,34 +88,60 @@ interface TaskStorage
/** /**
* Write the task output into TaskStorage. * Write the task output into TaskStorage.
* *
* @param Task $task * $attempt refers to $task->getRetries(). If 0, it is the initial attempt. If > 0, it seeks a retry output.
* @param string $output
* @param string $errors
* @param int $statusCode
* @return bool
*/
public function writeTaskOutput(Task $task, string $output, string $errors, int $statusCode): bool;
/**
* Write the output of the postHandler into TaskStorage
* *
* @param Task $task * @param Task $task
* @param string $output * @param string $output
* @param string $errors * @param string $errors
* @param int $statusCode * @param int $statusCode
* @param int $attempt
* @return bool * @return bool
* @throws TasksException
*/ */
public function writePostOutput(Task $task, string $output, string $errors, int $statusCode): bool; public function writeTaskOutput(Task $task, string $output, string $errors, int $statusCode, int $attempt = 0): bool;
/**
* Write the output of the postHandler into TaskStorage
*
* $attempt refers to $task->getRetries(). If 0, it is the initial attempt. If > 0, it seeks a retry output.
*
* @param Task $task
* @param string $output
* @param string $errors
* @param int $statusCode
* @param int $attempt
* @return bool
* @throws TasksException
*/
public function writePostOutput(Task $task, string $output, string $errors, int $statusCode, int $attempt = 0): bool;
/** /**
* Read the task output from taskStorage. * Read the task output from taskStorage.
* *
* The following output is expected: * The following output is expected:
* array('output' => string $output, 'errors' => string $errors, 'status' => $code) * array('output' => string $output, 'errors' => string $errors, 'status' => $code)
* OR null of not found (yet) * OR null if not found (yet)
*
* $attempt refers to $task->getRetries(). If 0, it is the initial attempt. If > 0, it seeks a retry output.
* *
* @param Task $task * @param Task $task
* @param int $attempt
* @return array * @return array
*/ */
public function readTaskOutput(Task $task): ?array; public function readTaskOutput(Task $task, int $attempt = 0): ?array;
/**
* Read the output from the postHandler
*
* The following output is expected:
* array('output' => string $output, 'errors' => string $errors, 'status' => $code)
* OR null if not found (yet)
*
* $attempt refers to $task->getRetries(). If 0, it is the initial attempt. If > 0, it seeks a retry output.
*
* @param Task $task
* @param int $attempt
* @return array|null
*/
public function readPostOutput(Task $task, int $attempt = 0): ?array;
} }

63
Tasks/src/TaskStorage/ArrayTaskStorage.php → src/FuzeWorks/Async/TaskStorage/ArrayTaskStorage.php
View File

@ -1,10 +1,10 @@
<?php <?php
/** /**
* FuzeWorks CLIComponent. * FuzeWorks Async Library
* *
* The FuzeWorks PHP FrameWork * The FuzeWorks PHP FrameWork
* *
* Copyright (C) 2013-2019 TechFuze * Copyright (C) 2013-2020 TechFuze
* *
* Permission is hereby granted, free of charge, to any person obtaining a copy * Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal * of this software and associated documentation files (the "Software"), to deal
@ -25,19 +25,19 @@
* SOFTWARE. * SOFTWARE.
* *
* @author TechFuze * @author TechFuze
* @copyright Copyright (c) 2013 - 2019, TechFuze. (http://techfuze.net) * @copyright Copyright (c) 2013 - 2020, TechFuze. (http://techfuze.net)
* @license https://opensource.org/licenses/MIT MIT License * @license https://opensource.org/licenses/MIT MIT License
* *
* @link http://techfuze.net/fuzeworks * @link http://techfuze.net/fuzeworks
* @since Version 1.2.0 * @since Version 1.0.0
* *
* @version Version 1.2.0 * @version Version 1.0.0
*/ */
namespace Application\Library\Tasks\TaskStorage; namespace FuzeWorks\Async\TaskStorage;
use Application\Library\Tasks\Task; use FuzeWorks\Async\Task;
use Application\Library\Tasks\TasksException; use FuzeWorks\Async\TasksException;
use Application\Library\Tasks\TaskStorage; use FuzeWorks\Async\TaskStorage;
class ArrayTaskStorage implements TaskStorage class ArrayTaskStorage implements TaskStorage
{ {
@ -184,11 +184,11 @@ class ArrayTaskStorage implements TaskStorage
* @inheritDoc * @inheritDoc
* @throws TasksException * @throws TasksException
*/ */
public function writeTaskOutput(Task $task, string $output, string $errors, int $statusCode): bool public function writeTaskOutput(Task $task, string $output, string $errors, int $statusCode, int $attempt = 0): bool
{ {
// Get the directory of the main storage // Get the directory of the main storage
$dir = dirname($this->fileName); $dir = dirname($this->fileName);
$file = $dir . DS . 'task_' . md5($task->getId()) . '_output.json'; $file = $dir . DS . 'task_' . md5($task->getId()) . '_' . $attempt . '_output.json';
$contents = json_encode(['taskId' => $task->getId(), 'output' => $output, 'errors' => $errors, 'statusCode' => $statusCode]); $contents = json_encode(['taskId' => $task->getId(), 'output' => $output, 'errors' => $errors, 'statusCode' => $statusCode]);
// If file already exists, panic // If file already exists, panic
@ -205,11 +205,11 @@ class ArrayTaskStorage implements TaskStorage
* @inheritDoc * @inheritDoc
* @throws TasksException * @throws TasksException
*/ */
public function writePostOutput(Task $task, string $output, string $errors, int $statusCode): bool public function writePostOutput(Task $task, string $output, string $errors, int $statusCode, int $attempt = 0): bool
{ {
// Get the directory of the main storage // Get the directory of the main storage
$dir = dirname($this->fileName); $dir = dirname($this->fileName);
$file = $dir . DS . 'task_' . md5($task->getId()) . '_post_output.json'; $file = $dir . DS . 'task_' . md5($task->getId()) . '_' . $attempt . '_post_output.json';
$contents = json_encode(['taskId' => $task->getId(), 'output' => $output, 'errors' => $errors, 'statusCode' => $statusCode]); $contents = json_encode(['taskId' => $task->getId(), 'output' => $output, 'errors' => $errors, 'statusCode' => $statusCode]);
// If file already exists, panic // If file already exists, panic
@ -225,12 +225,12 @@ class ArrayTaskStorage implements TaskStorage
/** /**
* @inheritDoc * @inheritDoc
*/ */
public function readTaskOutput(Task $task): ?array public function readTaskOutput(Task $task, int $attempt = 0): ?array
{ {
// Get the directory of the main storage // Get the directory of the main storage
$dir = dirname($this->fileName); $dir = dirname($this->fileName);
$file = $dir . DS . 'task_' . md5($task->getId()) . '_output.json'; $file = $dir . DS . 'task_' . md5($task->getId()) . '_' . $attempt . '_output.json';
$file2 = $dir . DS . 'task_' . md5($task->getId()) . '_post_output.json';
// If file doesn't exist, return so // If file doesn't exist, return so
if (!file_exists($file)) if (!file_exists($file))
@ -239,20 +239,27 @@ class ArrayTaskStorage implements TaskStorage
// Decode the contents // Decode the contents
$contents = file_get_contents($file); $contents = file_get_contents($file);
$data = json_decode($contents, true); $data = json_decode($contents, true);
$output = ['output' => $data['output'], 'errors' => $data['errors'], 'statusCode' => $data['statusCode']];
// Add post data if necessary return ['output' => $data['output'], 'errors' => $data['errors'], 'statusCode' => $data['statusCode']];
if (file_exists($file2)) }
{
$postContents = file_get_contents($file2);
$postData = json_decode($postContents, true);
$output['postOutput'] = $postData['output'];
$output['postErrors'] = $postData['errors'];
$output['postStatus'] = $postData['statusCode'];
}
// And return them /**
return $output; * @inheritDoc
*/
public function readPostOutput(Task $task, int $attempt = 0): ?array
{
// Get the directory of the main storage
$dir = dirname($this->fileName);
$file = $dir . DS . 'task_' . md5($task->getId()) . '_' . $attempt . '_post_output.json';
// If file doesn't exist, return so
if (!file_exists($file))
return null;
// Decode the contents
$contents = file_get_contents($file);
$data = json_decode($contents, true);
return ['output' => $data['output'], 'errors' => $data['errors'], 'statusCode' => $data['statusCode']];
} }
private function commit() private function commit()

31
Tasks/Tasks.php → src/FuzeWorks/Async/Tasks.php
View File

@ -1,10 +1,10 @@
<?php <?php
/** /**
* FuzeWorks CLIComponent. * FuzeWorks Async Library
* *
* The FuzeWorks PHP FrameWork * The FuzeWorks PHP FrameWork
* *
* Copyright (C) 2013-2019 TechFuze * Copyright (C) 2013-2020 TechFuze
* *
* Permission is hereby granted, free of charge, to any person obtaining a copy * Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal * of this software and associated documentation files (the "Software"), to deal
@ -25,25 +25,18 @@
* SOFTWARE. * SOFTWARE.
* *
* @author TechFuze * @author TechFuze
* @copyright Copyright (c) 2013 - 2019, TechFuze. (http://techfuze.net) * @copyright Copyright (c) 2013 - 2020, TechFuze. (http://techfuze.net)
* @license https://opensource.org/licenses/MIT MIT License * @license https://opensource.org/licenses/MIT MIT License
* *
* @link http://techfuze.net/fuzeworks * @link http://techfuze.net/fuzeworks
* @since Version 1.2.0 * @since Version 1.0.0
* *
* @version Version 1.2.0 * @version Version 1.0.0
*/ */
namespace Application\Library; namespace FuzeWorks\Async;
use Application\Library\Tasks\Executor;
use Application\Library\Tasks\SuperVisor;
use Application\Library\Tasks\Task;
use Application\Library\Tasks\TasksException;
use Application\Library\Tasks\TaskStorage;
use Application\Library\Tasks\Worker;
use FuzeWorks\Config; use FuzeWorks\Config;
use FuzeWorks\ConfigORM\ConfigORM; use FuzeWorks\ConfigORM\ConfigORM;
use FuzeWorks\Events;
use FuzeWorks\Exception\Exception; use FuzeWorks\Exception\Exception;
use FuzeWorks\Factory; use FuzeWorks\Factory;
use FuzeWorks\iLibrary; use FuzeWorks\iLibrary;
@ -66,7 +59,7 @@ class Tasks implements iLibrary
{ {
/** @var Config $config */ /** @var Config $config */
$config = Factory::getInstance('config'); $config = Factory::getInstance('config');
$config->addComponentPath(dirname(__FILE__), Priority::LOW); $config->addComponentPath(dirname(__FILE__, 4), Priority::LOW);
$this->cfg = $config->getConfig('tasks'); $this->cfg = $config->getConfig('tasks');
} }
@ -87,7 +80,7 @@ class Tasks implements iLibrary
public function getSuperVisor(): SuperVisor public function getSuperVisor(): SuperVisor
{ {
$cfg = $this->cfg->get('SuperVisor'); $cfg = $this->cfg->get('SuperVisor');
$class = 'Application\Library\Tasks\Supervisors\\' . $cfg['type']; $class = 'FuzeWorks\Async\Supervisors\\' . $cfg['type'];
$parameters = isset($cfg['parameters']) && is_array($cfg['parameters']) ? $cfg['parameters'] : []; $parameters = isset($cfg['parameters']) && is_array($cfg['parameters']) ? $cfg['parameters'] : [];
array_unshift($parameters, $this->getTaskStorage(), $this->getExecutor()); array_unshift($parameters, $this->getTaskStorage(), $this->getExecutor());
if (!class_exists($class, true)) if (!class_exists($class, true))
@ -118,7 +111,7 @@ class Tasks implements iLibrary
public function getTaskStorage(): TaskStorage public function getTaskStorage(): TaskStorage
{ {
$cfg = $this->cfg->get('TaskStorage'); $cfg = $this->cfg->get('TaskStorage');
$class = 'Application\Library\Tasks\TaskStorage\\' . $cfg['type']; $class = 'FuzeWorks\Async\TaskStorage\\' . $cfg['type'];
$parameters = isset($cfg['parameters']) && is_array($cfg['parameters']) ? $cfg['parameters'] : []; $parameters = isset($cfg['parameters']) && is_array($cfg['parameters']) ? $cfg['parameters'] : [];
if (!class_exists($class, true)) if (!class_exists($class, true))
throw new TasksException("Could not get TaskStorage. Type of '$class' not found."); throw new TasksException("Could not get TaskStorage. Type of '$class' not found.");
@ -139,7 +132,7 @@ class Tasks implements iLibrary
protected function getExecutor(): Executor protected function getExecutor(): Executor
{ {
$cfg = $this->cfg->get('Executor'); $cfg = $this->cfg->get('Executor');
$class = 'Application\Library\Tasks\Executors\\' . $cfg['type']; $class = 'FuzeWorks\Async\Executors\\' . $cfg['type'];
$parameters = isset($cfg['parameters']) && is_array($cfg['parameters']) ? $cfg['parameters'] : []; $parameters = isset($cfg['parameters']) && is_array($cfg['parameters']) ? $cfg['parameters'] : [];
if (!class_exists($class, true)) if (!class_exists($class, true))
throw new TasksException("Could not get Executor. Type of '$class' not found."); throw new TasksException("Could not get Executor. Type of '$class' not found.");
@ -156,7 +149,7 @@ class Tasks implements iLibrary
*/ */
public function getClassesPrefix(): ?string public function getClassesPrefix(): ?string
{ {
return '\Application\Library\Tasks'; return null;
} }
/** /**
@ -164,6 +157,6 @@ class Tasks implements iLibrary
*/ */
public function getSourceDirectory(): ?string public function getSourceDirectory(): ?string
{ {
return 'src'; return null;
} }
} }

12
Tasks/src/TasksException.php → src/FuzeWorks/Async/TasksException.php
View File

@ -1,10 +1,10 @@
<?php <?php
/** /**
* FuzeWorks CLIComponent. * FuzeWorks Async Library
* *
* The FuzeWorks PHP FrameWork * The FuzeWorks PHP FrameWork
* *
* Copyright (C) 2013-2019 TechFuze * Copyright (C) 2013-2020 TechFuze
* *
* Permission is hereby granted, free of charge, to any person obtaining a copy * Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal * of this software and associated documentation files (the "Software"), to deal
@ -25,16 +25,16 @@
* SOFTWARE. * SOFTWARE.
* *
* @author TechFuze * @author TechFuze
* @copyright Copyright (c) 2013 - 2019, TechFuze. (http://techfuze.net) * @copyright Copyright (c) 2013 - 2020, TechFuze. (http://techfuze.net)
* @license https://opensource.org/licenses/MIT MIT License * @license https://opensource.org/licenses/MIT MIT License
* *
* @link http://techfuze.net/fuzeworks * @link http://techfuze.net/fuzeworks
* @since Version 1.2.0 * @since Version 1.0.0
* *
* @version Version 1.2.0 * @version Version 1.0.0
*/ */
namespace Application\Library\Tasks; namespace FuzeWorks\Async;
use FuzeWorks\Exception\LibraryException; use FuzeWorks\Exception\LibraryException;
class TasksException extends LibraryException class TasksException extends LibraryException

74
Tasks/src/Worker.php → src/FuzeWorks/Async/Worker.php
View File

@ -1,10 +1,10 @@
<?php <?php
/** /**
* FuzeWorks CLIComponent. * FuzeWorks Async Library
* *
* The FuzeWorks PHP FrameWork * The FuzeWorks PHP FrameWork
* *
* Copyright (C) 2013-2019 TechFuze * Copyright (C) 2013-2020 TechFuze
* *
* Permission is hereby granted, free of charge, to any person obtaining a copy * Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal * of this software and associated documentation files (the "Software"), to deal
@ -25,17 +25,17 @@
* SOFTWARE. * SOFTWARE.
* *
* @author TechFuze * @author TechFuze
* @copyright Copyright (c) 2013 - 2019, TechFuze. (http://techfuze.net) * @copyright Copyright (c) 2013 - 2020, TechFuze. (http://techfuze.net)
* @license https://opensource.org/licenses/MIT MIT License * @license https://opensource.org/licenses/MIT MIT License
* *
* @link http://techfuze.net/fuzeworks * @link http://techfuze.net/fuzeworks
* @since Version 1.2.0 * @since Version 1.0.0
* *
* @version Version 1.2.0 * @version Version 1.0.0
*/ */
namespace Application\Library\Tasks; namespace FuzeWorks\Async;
use Application\Library\Tasks\Events\TaskHandleEvent; use FuzeWorks\Async\Events\TaskHandleEvent;
use FuzeWorks\Event\HaltExecutionEvent; use FuzeWorks\Event\HaltExecutionEvent;
use FuzeWorks\Events; use FuzeWorks\Events;
use FuzeWorks\Exception\EventException; use FuzeWorks\Exception\EventException;
@ -50,6 +50,16 @@ class Worker
*/ */
protected $taskStorage; protected $taskStorage;
/**
* @var Task
*/
protected $task;
/**
* @var bool
*/
protected $post;
public function __construct(TaskStorage $taskStorage) public function __construct(TaskStorage $taskStorage)
{ {
$this->taskStorage = $taskStorage; $this->taskStorage = $taskStorage;
@ -72,45 +82,65 @@ class Worker
$event = Events::fireEvent(new TaskHandleEvent(), $task); $event = Events::fireEvent(new TaskHandleEvent(), $task);
$task = $event->getTask(); $task = $event->getTask();
// Set task to this worker
$this->task = $task;
$this->post = $post;
// Fetch the callable // Fetch the callable
$class = $task->getHandlerClass(); $class = $this->task->getHandlerClass();
if (!class_exists($class, true)) if (!class_exists($class, true))
{
$errors = 'Could not run task. HandlerClass \'' . $class . '\' not found.';
if (!$post)
$this->taskStorage->writeTaskOutput($this->task, '', $errors, Task::PFAILED, $this->task->getRetries());
else
$this->taskStorage->writePostOutput($this->task, '', $errors, Task::PFAILED, $this->task->getRetries());
throw new TasksException("Could not run task. '$class' not found."); throw new TasksException("Could not run task. '$class' not found.");
}
// Create the handler // Create the handler
/** @var Handler $object */ /** @var Handler $object */
$object = new $class(); $object = new $class();
if (!$object instanceof Handler) if (!$object instanceof Handler)
{
$errors = "Could not run task. '$class' is not instance of Handler.";
if (!$post)
$this->taskStorage->writeTaskOutput($this->task, '', $errors, Task::PFAILED, $this->task->getRetries());
else
$this->taskStorage->writePostOutput($this->task, '', $errors, Task::PFAILED, $this->task->getRetries());
throw new TasksException("Could not run task. '$class' is not instance of Handler."); throw new TasksException("Could not run task. '$class' is not instance of Handler.");
}
// Run postHandler if post mode is requested // Run postHandler if post mode is requested
if ($post) if ($post)
{ {
$postSuccess = $object->postHandler($task); $postSuccess = $object->postHandler($this->task);
$postOutput = $object->getPostOutput(); $postOutput = $object->getPostOutput();
$postOutput = is_null($postOutput) ? '' : (string) $postOutput; $postOutput = is_null($postOutput) ? '' : (string) $postOutput;
$postErrors = $this->getErrors(); $postErrors = $this->getErrors();
if (!$postSuccess) if (!$postSuccess)
$this->taskStorage->writePostOutput($task, $postOutput, $postErrors, Task::FAILED); $this->taskStorage->writePostOutput($this->task, $postOutput, $postErrors, Task::FAILED, $this->task->getRetries());
else else
$this->taskStorage->writePostOutput($task, $postOutput, $postErrors, Task::SUCCESS); $this->taskStorage->writePostOutput($this->task, $postOutput, $postErrors, Task::SUCCESS, $this->task->getRetries());
$this->output($postOutput, $postErrors); $this->output($postOutput, $postErrors);
return; return;
} }
// And execute // Run primaryHandler if requested
$success = $object->primaryHandler($task); $success = $object->primaryHandler($this->task);
$output = $object->getOutput(); $output = $object->getOutput();
$output = is_null($output) ? '' : (string) $output; $output = is_null($output) ? '' : (string) $output;
$errors = $this->getErrors(); $errors = $this->getErrors();
// And afterwards write the results to the TaskStorage // And afterwards write the results to the TaskStorage
if (!$success) if (!$success)
$this->taskStorage->writeTaskOutput($task, $output, $errors, Task::FAILED); $this->taskStorage->writeTaskOutput($this->task, $output, $errors, Task::FAILED, $this->task->getRetries());
else else
$this->taskStorage->writeTaskOutput($task, $output, $errors, Task::SUCCESS); $this->taskStorage->writeTaskOutput($this->task, $output, $errors, Task::SUCCESS, $this->task->getRetries());
$this->output($output, $errors); $this->output($output, $errors);
} }
@ -128,6 +158,17 @@ class Worker
// Collect all error logs // Collect all error logs
$errors = $this->getErrors(); $errors = $this->getErrors();
$this->output('', $errors); $this->output('', $errors);
try {
// Write to TaskStorage
if (!$this->post)
$this->taskStorage->writeTaskOutput($this->task, '', $errors, Task::FAILED, $this->task->getRetries());
else
$this->taskStorage->writePostOutput($this->task, '', $errors, Task::FAILED, $this->task->getRetries());
} catch (TasksException $e) {
// Ignore
}
exit; exit;
} }
@ -145,8 +186,7 @@ class Worker
continue; continue;
$output[] = strtoupper($log['type']) . ' ' . $output[] = strtoupper($log['type']) . ' ' .
(!empty($log['logFile']) && !empty($log['logLine']) ? $log['logFile'] . ':' . $log['logLine'] . " '" : '') . (!empty($log['logFile']) && !empty($log['logLine']) ? $log['logFile'] . ':' . $log['logLine'] . " '" : "'") . $log['message'] . "'";
$log['message'];
} }
return implode("\n", $output); return implode("\n", $output);

0
Tasks/storage.php → storage.php
View File

70
supervisor.php Normal file
View File

@ -0,0 +1,70 @@
<?php
/**
* FuzeWorks Async Library
*
* The FuzeWorks PHP FrameWork
*
* Copyright (C) 2013-2020 TechFuze
*
* Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal
* in the Software without restriction, including without limitation the rights
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
* copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
* SOFTWARE.
*
* @author TechFuze
* @copyright Copyright (c) 2013 - 2020, TechFuze. (http://techfuze.net)
* @license https://opensource.org/licenses/MIT MIT License
*
* @link http://techfuze.net/fuzeworks
* @since Version 1.0.0
*
* @version Version 1.0.0
*/
use FuzeWorks\Async\SuperVisor;
use FuzeWorks\Async\Tasks;
use FuzeWorks\Logger;
require_once('vendor/autoload.php');
// Open configurator
$configurator = new FuzeWorks\Configurator();
// Set up basic settings
$configurator->setTimeZone('Europe/Amsterdam');
$configurator->setTempDirectory(dirname(__FILE__) . DIRECTORY_SEPARATOR . 'temp');
$configurator->setLogDirectory(dirname(__FILE__). DIRECTORY_SEPARATOR . 'log');
// Add Async library
$configurator->deferComponentClassMethod('libraries', 'addLibraryClass', null, 'async', '\FuzeWorks\Async\Tasks');
// Debug
$configurator->enableDebugMode()->setDebugAddress('ALL');
// Create container
$container = $configurator->createContainer();
// Add lib
Logger::enableScreenLog();
// RUN THE APP
/** @var Tasks $lib */
$lib = $container->libraries->get('async');
$supervisor = $lib->getSuperVisor();
while ($supervisor->cycle() === SuperVisor::RUNNING) {
usleep(250000);
}

42
Tasks/worker.php → worker.php
View File

@ -1,10 +1,10 @@
<?php <?php
/** /**
* FuzeWorks Application Skeleton. * FuzeWorks Async Library
* *
* The FuzeWorks PHP FrameWork * The FuzeWorks PHP FrameWork
* *
* Copyright (C) 2013-2019 TechFuze * Copyright (C) 2013-2020 TechFuze
* *
* Permission is hereby granted, free of charge, to any person obtaining a copy * Permission is hereby granted, free of charge, to any person obtaining a copy
* of this software and associated documentation files (the "Software"), to deal * of this software and associated documentation files (the "Software"), to deal
@ -25,48 +25,34 @@
* SOFTWARE. * SOFTWARE.
* *
* @author TechFuze * @author TechFuze
* @copyright Copyright (c) 2013 - 2019, TechFuze. (http://techfuze.net) * @copyright Copyright (c) 2013 - 2020, TechFuze. (http://techfuze.net)
* @license https://opensource.org/licenses/MIT MIT License * @license https://opensource.org/licenses/MIT MIT License
* *
* @link http://techfuze.net/fuzeworks * @link http://techfuze.net/fuzeworks
* @since Version 0.0.1 * @since Version 1.0.0
* *
* @version Version 1.2.0 * @version Version 1.0.0
*/ */
use FuzeWorks\Async\Tasks;
use Application\Library\Tasks; require_once('vendor/autoload.php');
use FuzeWorks\Priority;
use FuzeWorks\WebAppComponent;
require_once(dirname(__DIR__) . DIRECTORY_SEPARATOR . 'vendor/autoload.php');
// Open configurator // Open configurator
$configurator = new FuzeWorks\Configurator(); $configurator = new FuzeWorks\Configurator();
// Set up basic settings // Set up basic settings
$configurator->setTimeZone('Europe/Amsterdam'); $configurator->setTimeZone('Europe/Amsterdam');
$configurator->setTempDirectory(dirname(__DIR__) . DIRECTORY_SEPARATOR . 'temp'); $configurator->setTempDirectory(dirname(__FILE__) . DIRECTORY_SEPARATOR . 'temp');
$configurator->setLogDirectory(dirname(__DIR__). DIRECTORY_SEPARATOR . 'log'); $configurator->setLogDirectory(dirname(__FILE__). DIRECTORY_SEPARATOR . 'log');
// Enable components // Add Async library
// WebComponent $configurator->deferComponentClassMethod('libraries', 'addLibraryClass', null, 'async', '\FuzeWorks\Async\Tasks');
$webAppComponent = new WebAppComponent();
$configurator->addComponent($webAppComponent);
// Add directories
$configurator->addDirectory(dirname(__FILE__) . DIRECTORY_SEPARATOR . 'Config', 'config', Priority::HIGH);
$configurator->addDirectory(dirname(__FILE__) . DIRECTORY_SEPARATOR . 'Helper', 'helpers', Priority::HIGH);
$configurator->addDirectory(dirname(__FILE__) . DIRECTORY_SEPARATOR . 'Library', 'libraries', Priority::HIGH);
$configurator->addDirectory(dirname(__FILE__) . DIRECTORY_SEPARATOR . 'Plugin', 'plugins', Priority::HIGH);
$configurator->addDirectory(dirname(__FILE__) . DIRECTORY_SEPARATOR . 'Controller', 'controllers', Priority::HIGH);
$configurator->addDirectory(dirname(__FILE__) . DIRECTORY_SEPARATOR . 'View', 'views', Priority::HIGH);
$configurator->addDirectory(dirname(__FILE__) . DIRECTORY_SEPARATOR . 'Model', 'models', Priority::HIGH);
$configurator->addDirectory(dirname(__FILE__) . DIRECTORY_SEPARATOR . 'Layout', 'layouts', Priority::NORMAL);
// Debug // Debug
$configurator->enableDebugMode()->setDebugAddress('ALL'); $configurator->enableDebugMode()->setDebugAddress('ALL');
// Create container
$container = $configurator->createContainer(); $container = $configurator->createContainer();
//\FuzeWorks\Logger::enableScreenLog();
// Prepare arguments // Prepare arguments
$script = array_shift($argv); $script = array_shift($argv);
@ -77,6 +63,6 @@ $post = ($mode === 'post' ? true : false);
// RUN THE APP // RUN THE APP
/** @var Tasks $lib */ /** @var Tasks $lib */
$lib = $container->libraries->get('tasks'); $lib = $container->libraries->get('async');
$worker = $lib->getWorker(); $worker = $lib->getWorker();
$worker->run($taskID, $post); $worker->run($taskID, $post);