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

Release of RC1 #7

Merged
abelhooge merged 34 commits from 3-features into master 2020-06-07 13:54:20 +00:00
Conversation 0 Commits 34 Files Changed 39 +161 -43
6 changed files with 161 additions and 43 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit fc83a931ae - Show all commits

6
src/FuzeWorks/Async/Supervisors/ParallelSuperVisor.php
View File

@ -117,8 +117,9 @@ class ParallelSuperVisor implements SuperVisor
// RUNNING: check if task is still running. If not, set result based on output
elseif ($task->getStatus() === Task::RUNNING)
{
// @todo Find a way to use the latest output
$isRunning = $this->executor->getTaskRunning($task);
$output = $this->taskStorage->readTaskOutput($task);
$output = $this->taskStorage->readTaskOutput($task, 1);
$hasOutput = !is_null($output);
// If nothing is found, the process has crashed and status PFAILED should be set
@ -203,8 +204,9 @@ class ParallelSuperVisor implements SuperVisor
// POST: when a task is currently running in it's postHandler
elseif ($task->getStatus() === Task::POST)
{
// @todo Find a way to use the latest output
$isRunning = $this->executor->getTaskRunning($task);
$output = $this->taskStorage->readPostOutput($task);
$output = $this->taskStorage->readPostOutput($task, 1);
$hasOutput = !is_null($output);
// If a task is not running and has no output, an error has occurred

10
src/FuzeWorks/Async/TaskStorage.php
View File

@ -135,7 +135,8 @@ interface TaskStorage
* 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.
* Attempt refers to the attempt that the task has ran and that individual output.
* If > 0, the output as mentioned above will be returned. If = 0, an array of such outputs will be returned.
*
* Returns null because that is a very valid response. Oftentimes output will need to be checked and its undesirable
* to always throw an exception for expected behaviour.
@ -144,7 +145,7 @@ interface TaskStorage
* @param int $attempt
* @return array|null
*/
public function readTaskOutput(Task $task, int $attempt = 1): ?array;
public function readTaskOutput(Task $task, int $attempt = 0): ?array;
/**
* Read the output from the postHandler
@ -153,7 +154,8 @@ interface TaskStorage
* 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.
* Attempt refers to the attempt that the task has ran and that individual output.
* If > 0, the output as mentioned above will be returned. If = 0, an array of such outputs will be returned.
*
* Returns null because that is a very valid response. Oftentimes output will need to be checked and its undesirable
* to always throw an exception for expected behaviour.
@ -162,7 +164,7 @@ interface TaskStorage
* @param int $attempt
* @return array|null
*/
public function readPostOutput(Task $task, int $attempt = 1): ?array;
public function readPostOutput(Task $task, int $attempt = 0): ?array;
/**
* Reset the TaskStorage.

28
src/FuzeWorks/Async/TaskStorage/DummyTaskStorage.php
View File

@ -235,10 +235,18 @@ class DummyTaskStorage implements TaskStorage
/**
* @inheritDoc
*/
public function readTaskOutput(Task $task, int $attempt = 1): ?array
public function readTaskOutput(Task $task, int $attempt = 0): ?array
{
if (isset($this->taskOutput[$task->getId()]['task'][$attempt]))
return $this->taskOutput[$task->getId()]['task'][$attempt];
if ($attempt !== 0)
{
if (isset($this->taskOutput[$task->getId()]['task'][$attempt]))
return $this->taskOutput[$task->getId()]['task'][$attempt];
}
else
{
if (isset($this->taskOutput[$task->getId()]['task']))
return $this->taskOutput[$task->getId()]['task'];
}
return null;
}
@ -246,10 +254,18 @@ class DummyTaskStorage implements TaskStorage
/**
* @inheritDoc
*/
public function readPostOutput(Task $task, int $attempt = 1): ?array
public function readPostOutput(Task $task, int $attempt = 0): ?array
{
if (isset($this->taskOutput[$task->getId()]['post'][$attempt]))
return $this->taskOutput[$task->getId()]['post'][$attempt];
if ($attempt !== 0)
{
if (isset($this->taskOutput[$task->getId()]['post'][$attempt]))
return $this->taskOutput[$task->getId()]['post'][$attempt];
}
else
{
if (isset($this->taskOutput[$task->getId()]['post']))
return $this->taskOutput[$task->getId()]['post'];
}
return null;
}

88
src/FuzeWorks/Async/TaskStorage/RedisTaskStorage.php
View File

@ -266,39 +266,91 @@ class RedisTaskStorage implements TaskStorage
/**
* @inheritDoc
*/
public function readTaskOutput(Task $task, int $attempt = 1): ?array
public function readTaskOutput(Task $task, int $attempt = 0): ?array
{
// First get the task ID
$taskId = $task->getId();
// Check if this output exists
if (!$this->conn->hExists($this->key_prefix . $taskId, 'output' . $attempt))
if ($attempt !== 0)
{
// Check if this output exists
if (!$this->conn->hExists($this->key_prefix . $taskId, 'output' . $attempt))
return null;
// Load and convert the data
$data = $this->conn->hGet($this->key_prefix . $taskId, 'output' . $attempt);
$data = unserialize($data);
return ['output' => $data['output'], 'errors' => $data['errors'], 'statusCode' => $data['statusCode']];
}
else
{
// Get amount of attempts
$totalAttempts = $this->conn->hGet($this->key_prefix . $taskId, 'taskOutputAttempts');
$output = [];
for ($i=1;$i<=$totalAttempts;$i++)
{
// Check if this output exists
if (!$this->conn->hExists($this->key_prefix . $taskId, 'output' . $i))
$output[] = null;
// Load and convert the data
$data = $this->conn->hGet($this->key_prefix . $taskId, 'output' . $i);
$data = unserialize($data);
$output[] = ['output' => $data['output'], 'errors' => $data['errors'], 'statusCode' => $data['statusCode']];
}
if (!empty($output))
return $output;
return null;
// Load and convert the data
$data = $this->conn->hGet($this->key_prefix . $taskId, 'output' . $attempt);
$data = unserialize($data);
return ['output' => $data['output'], 'errors' => $data['errors'], 'statusCode' => $data['statusCode']];
}
}
/**
* @inheritDoc
*/
public function readPostOutput(Task $task, int $attempt = 1): ?array
public function readPostOutput(Task $task, int $attempt = 0): ?array
{
// First get the task ID
$taskId = $task->getId();
// Check if this output exists
if (!$this->conn->hExists($this->key_prefix . $taskId, 'postOutput' . $attempt))
if ($attempt !== 0)
{
// Check if this output exists
if (!$this->conn->hExists($this->key_prefix . $taskId, 'postOutput' . $attempt))
return null;
// Load and convert the data
$data = $this->conn->hGet($this->key_prefix . $taskId, 'postOutput' . $attempt);
$data = unserialize($data);
return ['output' => $data['output'], 'errors' => $data['errors'], 'statusCode' => $data['statusCode']];
}
else
{
// Get amount of attempts
$totalAttempts = $this->conn->hGet($this->key_prefix . $taskId, 'taskPostAttempts');
$output = [];
for ($i=1;$i<=$totalAttempts;$i++)
{
// Check if this output exists
if (!$this->conn->hExists($this->key_prefix . $taskId, 'postOutput' . $i))
$output[] = null;
// Load and convert the data
$data = $this->conn->hGet($this->key_prefix . $taskId, 'postOutput' . $i);
$data = unserialize($data);
$output[] = ['output' => $data['output'], 'errors' => $data['errors'], 'statusCode' => $data['statusCode']];
}
if (!empty($output))
return $output;
return null;
// Load and convert the data
$data = $this->conn->hGet($this->key_prefix . $taskId, 'postOutput' . $attempt);
$data = unserialize($data);
return ['output' => $data['output'], 'errors' => $data['errors'], 'statusCode' => $data['statusCode']];
}
}
/**

12
test/base/ShellWorkerTest.php
View File

@ -102,13 +102,13 @@ class ShellWorkerTest extends TestCase
$this->shellWorker->run($dummyTask, false);
// And verify if the Output is correctly set
$output = $this->taskStorage->readTaskOutput($dummyTask);
$output = $this->taskStorage->readTaskOutput($dummyTask, 1);
$this->assertEquals('Some Output!', $output['output']);
$this->assertEquals(Task::SUCCESS, $output['statusCode']);
// And run the post handler
$this->shellWorker->run($dummyTask, true);
$output = $this->taskStorage->readPostOutput($dummyTask);
$output = $this->taskStorage->readPostOutput($dummyTask, 1);
$this->assertEquals('Post Output!', $output['output']);
$this->assertEquals(Task::SUCCESS, $output['statusCode']);
}
@ -136,13 +136,13 @@ class ShellWorkerTest extends TestCase
$this->shellWorker->run($dummyTask, false);
// And verify if the Output is correctly set
$output = $this->taskStorage->readTaskOutput($dummyTask);
$output = $this->taskStorage->readTaskOutput($dummyTask, 1);
$this->assertEquals('', $output['output']);
$this->assertEquals(Task::FAILED, $output['statusCode']);
// And run a post failure
$this->shellWorker->run($dummyTask, true);
$output = $this->taskStorage->readPostOutput($dummyTask);
$output = $this->taskStorage->readPostOutput($dummyTask, 1);
$this->assertEquals('', $output['output']);
$this->assertEquals(Task::FAILED, $output['statusCode']);
}
@ -191,7 +191,7 @@ class ShellWorkerTest extends TestCase
$this->shellWorker->run($dummyTask, false);
// And verify if the Output is correctly set
$output = $this->taskStorage->readTaskOutput($dummyTask);
$output = $this->taskStorage->readTaskOutput($dummyTask, 1);
$this->assertEquals('Child Output', $output['output']);
$this->assertEquals(Task::SUCCESS, $output['statusCode']);
}
@ -232,7 +232,7 @@ class ShellWorkerTest extends TestCase
$this->shellWorker->run($dummyTask, false);
// And verify whether the task has indeed failed
$output = $this->taskStorage->readTaskOutput($dummyTask);
$output = $this->taskStorage->readTaskOutput($dummyTask, 1);
$this->assertEquals('Task failed successfully', $output['output']);
$this->assertEquals(Task::FAILED, $output['statusCode']);
}

60
test/base/TaskStorageTest.php
View File

@ -324,6 +324,27 @@ class TaskStorageTest extends TestCase
$this->assertEquals(0, $output['statusCode']);
}
/**
* @depends testWriteAndReadTaskOutput
*/
public function testWriteAndReadMultipleOutput()
{
// Prepare a dummy task
$dummyTask = new Task('testWriteAndReadMultipleOutput', new EmptyHandler());
// Write some task output
$this->taskStorage->addTask($dummyTask);
$this->assertTrue($this->taskStorage->writeTaskOutput($dummyTask, 'output1', 'errors1', 0));
$this->assertTrue($this->taskStorage->writeTaskOutput($dummyTask, 'output2', 'errors2', 0));
// Then try to read all the output
$output = $this->taskStorage->readTaskOutput($dummyTask);
$this->assertEquals([
['output' => 'output1', 'errors' => 'errors1', 'statusCode' => 0],
['output' => 'output2', 'errors' => 'errors2', 'statusCode' => 0]
], $output);
}
/**
* @depends testWriteAndReadTaskOutput
*/
@ -334,7 +355,7 @@ class TaskStorageTest extends TestCase
// Write output while the task does not exist yet, expect exception
$this->expectException(TasksException::class);
$this->taskStorage->writeTaskOutput($dummyTask, 'output', 'errors', 0, 0);
$this->taskStorage->writeTaskOutput($dummyTask, 'output', 'errors', 0);
}
/**
@ -371,9 +392,11 @@ class TaskStorageTest extends TestCase
// Attempt to load the default output
$output = $this->taskStorage->readTaskOutput($dummyTask);
$this->assertEquals('output0', $output['output']);
$this->assertEquals('errors0', $output['errors']);
$this->assertEquals(100, $output['statusCode']);
$this->assertEquals([
['output' => 'output0', 'errors' => 'errors0', 'statusCode' => 100],
['output' => 'output1', 'errors' => 'errors1', 'statusCode' => 101],
['output' => 'output2', 'errors' => 'errors2', 'statusCode' => 102]
], $output);
}
/**
@ -409,6 +432,27 @@ class TaskStorageTest extends TestCase
$this->assertEquals(0, $output['statusCode']);
}
/**
* @depends testWriteAndReadTaskOutput
*/
public function testWriteAndReadMultiplePostOutput()
{
// Prepare a dummy task
$dummyTask = new Task('testWriteAndReadMultiplePostOutput', new EmptyHandler());
// Write some task output
$this->taskStorage->addTask($dummyTask);
$this->assertTrue($this->taskStorage->writePostOutput($dummyTask, 'output1', 'errors1', 0));
$this->assertTrue($this->taskStorage->writePostOutput($dummyTask, 'output2', 'errors2', 0));
// Then try to read all the output
$output = $this->taskStorage->readPostOutput($dummyTask);
$this->assertEquals([
['output' => 'output1', 'errors' => 'errors1', 'statusCode' => 0],
['output' => 'output2', 'errors' => 'errors2', 'statusCode' => 0]
], $output);
}
/**
* @depends testWriteAndReadTaskPostOutput
*/
@ -444,9 +488,11 @@ class TaskStorageTest extends TestCase
// Attempt to load the default output
$output = $this->taskStorage->readPostOutput($dummyTask);
$this->assertEquals('output0', $output['output']);
$this->assertEquals('errors0', $output['errors']);
$this->assertEquals(100, $output['statusCode']);
$this->assertEquals([
['output' => 'output0', 'errors' => 'errors0', 'statusCode' => 100],
['output' => 'output1', 'errors' => 'errors1', 'statusCode' => 101],
['output' => 'output2', 'errors' => 'errors2', 'statusCode' => 102]
], $output);
}
/**