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

Try with only Redis.
continuous-integration/drone/push Build is passing Details

This commit is contained in:
Abel Hoogeveen 2020-05-26 17:56:15 +02:00
parent 5f5718cb72
commit d9da4cfc95
No known key found for this signature in database
GPG Key ID: 96C2234920BF4292
6 changed files with 120 additions and 47 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
.drone.yml
View File

@ -12,13 +12,6 @@ steps:
commands:
- composer install
- name: basetest
image: phpunit:7.3
commands:
- vendor/bin/phpunit -c test/phpunit.xml --coverage-php test/temp/covbase.cov
environment:
TASKSTORAGE_TYPE: DummyTaskStorage
- name: redistest
image: phpunit:7.3
commands:

50
src/FuzeWorks/Async/TaskStorage/ArrayTaskStorage.php
View File

@ -168,10 +168,20 @@ class ArrayTaskStorage implements TaskStorage
unset($this->tasks[$i]);
$this->commit();
// Afterwards remove the task output, if it exists
$file = dirname($this->fileName) . DS . 'task_' . md5($taskId) . '_output.json';
if (file_exists($file))
unlink($file);
// Remove all task output and post output
$settings = $task->getRetrySettings();
$maxRetries = $settings['maxRetries'];
for ($j=0;$j<=$maxRetries;$j++)
{
// First remove all possible task output
$outFile = dirname($this->fileName) . DS . 'task_' . md5($taskId) . '_' . $j . '_output.json';
if (file_exists($outFile))
unlink($outFile);
$postFile = dirname($this->fileName) . DS . 'task_' . md5($taskId) . '_' . $j . '_post_output.json';
if (file_exists($postFile))
unlink($postFile);
}
return true;
}
@ -264,11 +274,39 @@ class ArrayTaskStorage implements TaskStorage
/**
* @inheritDoc
* @throws TasksException
*/
public function reset(): bool
{
// @todo Implement
return false;
// Delete everything
$this->refreshTasks();
for ($i=0;$i<count($this->tasks);$i++)
{
// Get the task
$task = $this->tasks[$i];
// Remove all task output and post output
$settings = $task->getRetrySettings();
$maxRetries = $settings['maxRetries'];
for ($j=0;$j<=$maxRetries;$j++)
{
// First remove all possible task output
$outFile = dirname($this->fileName) . DS . 'task_' . md5($taskId) . '_' . $j . '_output.json';
if (file_exists($outFile))
unlink($outFile);
$postFile = dirname($this->fileName) . DS . 'task_' . md5($taskId) . '_' . $j . '_post_output.json';
if (file_exists($postFile))
unlink($postFile);
}
// Remove the task from the main storage
unset($this->tasks[$i]);
}
// And finally commit
$this->commit();
return true;
}
private function commit()

47
src/FuzeWorks/Async/Tasks.php
View File

@ -80,13 +80,26 @@ class Tasks implements iLibrary
*/
public function getSuperVisor(string $bootstrapFile): SuperVisor
{
// First get the configuration for SuperVisors
$cfg = $this->cfg->get('SuperVisor');
$class = 'FuzeWorks\Async\Supervisors\\' . $cfg['type'];
$parameters = isset($cfg['parameters']) && is_array($cfg['parameters']) ? $cfg['parameters'] : [];
// Select the SuperVisor type
$type = $cfg['type'];
// Load the class of the currently selected type
$class = 'FuzeWorks\Async\Supervisors\\' . $type;
// Fetch the parameters for the selected SuperVisor
$parameters = isset($cfg[$type]['parameters']) && is_array($cfg[$type]['parameters']) ? $cfg[$type]['parameters'] : [];
// Then add the TaskStorage and Executor to the parameters
array_unshift($parameters, $this->getTaskStorage(), $this->getExecutor($bootstrapFile));
// If the type does not exist, throw an exception
if (!class_exists($class, true))
throw new TasksException("Could not get SuperVisor. Type of '$class' not found.");
// And load the SuperVisor and test if everything is in order
$object = new $class(...$parameters);
if (!$object instanceof SuperVisor)
throw new TasksException("Could not get SuperVisor. Type of '$class' is not instanceof TaskStorage.");
@ -111,12 +124,23 @@ class Tasks implements iLibrary
*/
public function getTaskStorage(): TaskStorage
{
// First get the configuration for TaskStorage
$cfg = $this->cfg->get('TaskStorage');
$class = 'FuzeWorks\Async\TaskStorage\\' . $cfg['type'];
$parameters = isset($cfg['parameters']) && is_array($cfg['parameters']) ? $cfg['parameters'] : [];
// Select the TaskStorage type
$type = $cfg['type'];
// Load the class of the currently selected type
$class = 'FuzeWorks\Async\TaskStorage\\' . $type;
// Fetch the parameters for the selected type
$parameters = isset($cfg[$type]['parameters']) && is_array($cfg[$type]['parameters']) ? $cfg[$type]['parameters'] : [];
// If the type does not exist, throw an exception
if (!class_exists($class, true))
throw new TasksException("Could not get TaskStorage. Type of '$class' not found.");
// And load the TaskStorage and test if everything is in order
$object = new $class($parameters);
if (!$object instanceof TaskStorage)
throw new TasksException("Could not get TaskStorage. Type '$class' is not instanceof TaskStorage.");
@ -133,12 +157,23 @@ class Tasks implements iLibrary
*/
protected function getExecutor(string $bootstrapFile): Executor
{
// First get the configuration for Executor
$cfg = $this->cfg->get('Executor');
$class = 'FuzeWorks\Async\Executors\\' . $cfg['type'];
$parameters = isset($cfg['parameters']) && is_array($cfg['parameters']) ? $cfg['parameters'] : [];
// Select the Executor type
$type = $cfg['type'];
// Load the class of the currently selected type
$class = 'FuzeWorks\Async\Executors\\' . $type;
// Fetch the parameters for the selected type
$parameters = isset($cfg[$type]['parameters']) && is_array($cfg[$type]['parameters']) ? $cfg[$type]['parameters'] : [];
// If the type does not exist, throw an exception
if (!class_exists($class, true))
throw new TasksException("Could not get Executor. Type of '$class' not found.");
// And load the Executor and test if everything is in order
$object = new $class($bootstrapFile, $parameters);
if (!$object instanceof Executor)
throw new TasksException("Could not get Executor. Type '$class' is not instanceof Executor.");

61
test/config.tasks.php
View File

@ -43,42 +43,49 @@ return array(
// Add a file lock
'lock' => true,
// Which SuperVisor should be used
// Which SuperVisor should be used, and with what settings
'SuperVisor' => [
'type' => Core::getEnv('SUPERVISOR_TYPE', 'ParallelSuperVisor'),
'parameters' => []
'ParallelSuperVisor' => ['parameters' => []]
],
// Which TaskStorage should be used, and with what settings
'TaskStorage' => [
'type' => Core::getEnv('TASKSTORAGE_TYPE', 'DummyTaskStorage'),
// For ArrayTaskStorage, first parameter is the file location of the array storage
#'parameters' => [
# 'filename' => dirname(__FILE__) . DS . 'storage.php'
#],
// For RedisTaskStorage, parameters are connection properties
'parameters' => [
// Type can be 'tcp' or 'unix'
'socket_type' => Core::getEnv('TASKSTORAGE_REDIS_SOCKET_TYPE', 'tcp'),
// If socket_type == 'unix', set the socket here
'socket' => Core::getEnv('TASKSTORAGE_REDIS_SOCKET', null),
// If socket_type == 'tcp', set the host here
'host' => Core::getEnv('TASKSTORAGE_REDIS_HOST', '127.0.0.1'),
// And some standard settings
'password' => Core::getEnv('TASKSTORAGE_REDIS_PASSWORD', null),
'port' => Core::getEnv('TASKSTORAGE_REDIS_PORT', 6379),
'timeout' => Core::getEnv('TASKSTORAGE_REDIS_TIMEOUT', 0),
'db_index' => Core::getEnv('TASKSTORAGE_REDIS_DBINDEX', 0),
]
'DummyTaskStorage' => ['parameters' => []],
'ArrayTaskStorage' => [
'parameters' => [
'filename' => Core::getEnv('TASKSTORAGE_ARRAY_FILE',
dirname(__FILE__) . DS . 'temp'. DS . 'storage.php')
]
],
'RedisTaskStorage' => [
'parameters' => [
// Type can be 'tcp' or 'unix'
'socket_type' => Core::getEnv('TASKSTORAGE_REDIS_SOCKET_TYPE', 'tcp'),
// If socket_type == 'unix', set the socket here
'socket' => Core::getEnv('TASKSTORAGE_REDIS_SOCKET', null),
// If socket_type == 'tcp', set the host here
'host' => Core::getEnv('TASKSTORAGE_REDIS_HOST', '127.0.0.1'),
// And some standard settings
'password' => Core::getEnv('TASKSTORAGE_REDIS_PASSWORD', null),
'port' => Core::getEnv('TASKSTORAGE_REDIS_PORT', 6379),
'timeout' => Core::getEnv('TASKSTORAGE_REDIS_TIMEOUT', 0),
'db_index' => Core::getEnv('TASKSTORAGE_REDIS_DBINDEX', 0),
]
],
],
// Which Executor should be used, and with what settings
'Executor' => [
'type' => Core::getEnv('EXECUTOR_TYPE', 'ShellExecutor'),
// For ShellExecutor, first parameter is the file location of the worker script
'parameters' => [
'workerFile' => Core::getEnv('EXECUTOR_SHELL_WORKER',
dirname(__FILE__) . DS . 'bin' . DS . 'worker'),
'bootstrapFile' => Core::getEnv('EXECUTOR_SHELL_BOOTSTRAP', 'unknown')
'ShellExecutor' => [
'parameters' => [
'workerFile' => Core::getEnv('EXECUTOR_SHELL_WORKER',
dirname(__FILE__) . DS . 'bin' . DS . 'worker'),
'bootstrapFile' => Core::getEnv('EXECUTOR_SHELL_BOOTSTRAP', 'unknown')
]
]
]
);

2
test/mock/Handlers/TestStartAndReadTasksHandler.php
View File

@ -46,7 +46,7 @@ class TestStartAndReadTasksHandler implements Handler
*/
public function primaryHandler(Task $task): bool
{
sleep(2);
sleep(10);
return true;
}

0
storage.php → test/temp/storage.php
View File