More Laravel-esq job processing
This commit is contained in:
parent
e1e079bc12
commit
0f4f2235a3
3 changed files with 27 additions and 278 deletions
|
@ -3,10 +3,10 @@
|
|||
namespace Pterodactyl\Jobs\Schedule;
|
||||
|
||||
use Exception;
|
||||
use Carbon\Carbon;
|
||||
use Pterodactyl\Jobs\Job;
|
||||
use Carbon\CarbonImmutable;
|
||||
use Pterodactyl\Models\Task;
|
||||
use InvalidArgumentException;
|
||||
use Illuminate\Container\Container;
|
||||
use Illuminate\Queue\SerializesModels;
|
||||
use Illuminate\Queue\InteractsWithQueue;
|
||||
use Illuminate\Contracts\Queue\ShouldQueue;
|
||||
|
@ -15,39 +15,25 @@ use Pterodactyl\Repositories\Eloquent\TaskRepository;
|
|||
use Pterodactyl\Services\Backups\InitiateBackupService;
|
||||
use Pterodactyl\Repositories\Wings\DaemonPowerRepository;
|
||||
use Pterodactyl\Repositories\Wings\DaemonCommandRepository;
|
||||
use Pterodactyl\Contracts\Repository\TaskRepositoryInterface;
|
||||
use Pterodactyl\Contracts\Repository\ScheduleRepositoryInterface;
|
||||
|
||||
class RunTaskJob extends Job implements ShouldQueue
|
||||
{
|
||||
use DispatchesJobs, InteractsWithQueue, SerializesModels;
|
||||
|
||||
/**
|
||||
* @var int
|
||||
*/
|
||||
public $schedule;
|
||||
|
||||
/**
|
||||
* @var int
|
||||
* @var \Pterodactyl\Models\Task
|
||||
*/
|
||||
public $task;
|
||||
|
||||
/**
|
||||
* @var \Pterodactyl\Repositories\Eloquent\TaskRepository
|
||||
*/
|
||||
protected $taskRepository;
|
||||
|
||||
/**
|
||||
* RunTaskJob constructor.
|
||||
*
|
||||
* @param int $task
|
||||
* @param int $schedule
|
||||
* @param \Pterodactyl\Models\Task $task
|
||||
*/
|
||||
public function __construct(int $task, int $schedule)
|
||||
public function __construct(Task $task)
|
||||
{
|
||||
$this->queue = config('pterodactyl.queues.standard');
|
||||
$this->task = $task;
|
||||
$this->schedule = $schedule;
|
||||
}
|
||||
|
||||
/**
|
||||
|
@ -58,7 +44,6 @@ class RunTaskJob extends Job implements ShouldQueue
|
|||
* @param \Pterodactyl\Repositories\Wings\DaemonPowerRepository $powerRepository
|
||||
* @param \Pterodactyl\Repositories\Eloquent\TaskRepository $taskRepository
|
||||
*
|
||||
* @throws \Pterodactyl\Exceptions\Model\DataValidationException
|
||||
* @throws \Throwable
|
||||
*/
|
||||
public function handle(
|
||||
|
@ -67,36 +52,32 @@ class RunTaskJob extends Job implements ShouldQueue
|
|||
DaemonPowerRepository $powerRepository,
|
||||
TaskRepository $taskRepository
|
||||
) {
|
||||
$this->taskRepository = $taskRepository;
|
||||
|
||||
$task = $this->taskRepository->getTaskForJobProcess($this->task);
|
||||
$server = $task->getRelation('server');
|
||||
|
||||
// Do not process a task that is not set to active.
|
||||
if (! $task->getRelation('schedule')->is_active) {
|
||||
if (! $this->task->schedule->is_active) {
|
||||
$this->markTaskNotQueued();
|
||||
$this->markScheduleComplete();
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
$server = $this->task->server;
|
||||
// Perform the provided task against the daemon.
|
||||
switch ($task->action) {
|
||||
switch ($this->task->action) {
|
||||
case 'power':
|
||||
$powerRepository->setServer($server)->send($task->payload);
|
||||
$powerRepository->setServer($server)->send($this->task->payload);
|
||||
break;
|
||||
case 'command':
|
||||
$commandRepository->setServer($server)->send($task->payload);
|
||||
$commandRepository->setServer($server)->send($this->task->payload);
|
||||
break;
|
||||
case 'backup':
|
||||
$backupService->setIgnoredFiles(explode(PHP_EOL, $task->payload))->handle($server, null);
|
||||
$backupService->setIgnoredFiles(explode(PHP_EOL, $this->task->payload))->handle($server, null);
|
||||
break;
|
||||
default:
|
||||
throw new InvalidArgumentException('Cannot run a task that points to a non-existent action.');
|
||||
}
|
||||
|
||||
$this->markTaskNotQueued();
|
||||
$this->queueNextTask($task->sequence_id);
|
||||
$this->queueNextTask();
|
||||
}
|
||||
|
||||
/**
|
||||
|
@ -112,23 +93,23 @@ class RunTaskJob extends Job implements ShouldQueue
|
|||
|
||||
/**
|
||||
* Get the next task in the schedule and queue it for running after the defined period of wait time.
|
||||
*
|
||||
* @param int $sequence
|
||||
*
|
||||
* @throws \Pterodactyl\Exceptions\Model\DataValidationException
|
||||
* @throws \Pterodactyl\Exceptions\Repository\RecordNotFoundException
|
||||
*/
|
||||
private function queueNextTask($sequence)
|
||||
private function queueNextTask()
|
||||
{
|
||||
$nextTask = $this->taskRepository->getNextTask($this->schedule, $sequence);
|
||||
/** @var \Pterodactyl\Models\Task|null $nextTask */
|
||||
$nextTask = Task::query()->where('schedule_id', $this->task->schedule_id)
|
||||
->where('sequence_id', $this->task->sequence_id + 1)
|
||||
->first();
|
||||
|
||||
if (is_null($nextTask)) {
|
||||
$this->markScheduleComplete();
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
$this->taskRepository->update($nextTask->id, ['is_queued' => true]);
|
||||
$this->dispatch((new self($nextTask->id, $this->schedule))->delay($nextTask->time_offset));
|
||||
$nextTask->update(['is_queued' => true]);
|
||||
|
||||
$this->dispatch((new self($nextTask))->delay($nextTask->time_offset));
|
||||
}
|
||||
|
||||
/**
|
||||
|
@ -136,13 +117,10 @@ class RunTaskJob extends Job implements ShouldQueue
|
|||
*/
|
||||
private function markScheduleComplete()
|
||||
{
|
||||
Container::getInstance()
|
||||
->make(ScheduleRepositoryInterface::class)
|
||||
->withoutFreshModel()
|
||||
->update($this->schedule, [
|
||||
'is_processing' => false,
|
||||
'last_run_at' => Carbon::now()->toDateTimeString(),
|
||||
]);
|
||||
$this->task->schedule()->update([
|
||||
'is_processing' => false,
|
||||
'last_run_at' => CarbonImmutable::now()->toDateTimeString(),
|
||||
]);
|
||||
}
|
||||
|
||||
/**
|
||||
|
@ -150,8 +128,6 @@ class RunTaskJob extends Job implements ShouldQueue
|
|||
*/
|
||||
private function markTaskNotQueued()
|
||||
{
|
||||
Container::getInstance()
|
||||
->make(TaskRepositoryInterface::class)
|
||||
->update($this->task, ['is_queued' => false]);
|
||||
$this->task->update(['is_queued' => false]);
|
||||
}
|
||||
}
|
||||
|
|
|
@ -73,7 +73,7 @@ class ProcessScheduleService
|
|||
$this->taskRepository->update($task->id, ['is_queued' => true]);
|
||||
|
||||
$this->dispatcher->dispatch(
|
||||
(new RunTaskJob($task->id, $schedule->id))->delay($task->time_offset)
|
||||
(new RunTaskJob($task))->delay($task->time_offset)
|
||||
);
|
||||
}
|
||||
}
|
||||
|
|
|
@ -1,227 +0,0 @@
|
|||
<?php
|
||||
|
||||
namespace Tests\Unit\Jobs\Schedule;
|
||||
|
||||
use Mockery as m;
|
||||
use Carbon\Carbon;
|
||||
use Tests\TestCase;
|
||||
use Cake\Chronos\Chronos;
|
||||
use Pterodactyl\Models\Task;
|
||||
use Pterodactyl\Models\User;
|
||||
use GuzzleHttp\Psr7\Response;
|
||||
use InvalidArgumentException;
|
||||
use Pterodactyl\Models\Server;
|
||||
use Pterodactyl\Models\Schedule;
|
||||
use Illuminate\Support\Facades\Bus;
|
||||
use Pterodactyl\Jobs\Schedule\RunTaskJob;
|
||||
use Pterodactyl\Repositories\Eloquent\TaskRepository;
|
||||
use Pterodactyl\Services\Backups\InitiateBackupService;
|
||||
use Pterodactyl\Repositories\Eloquent\ScheduleRepository;
|
||||
use Pterodactyl\Repositories\Wings\DaemonPowerRepository;
|
||||
use Pterodactyl\Repositories\Wings\DaemonCommandRepository;
|
||||
use Pterodactyl\Contracts\Repository\TaskRepositoryInterface;
|
||||
use Pterodactyl\Contracts\Repository\ScheduleRepositoryInterface;
|
||||
|
||||
class RunTaskJobTest extends TestCase
|
||||
{
|
||||
/**
|
||||
* @var \Mockery\MockInterface
|
||||
*/
|
||||
private $commandRepository;
|
||||
|
||||
/**
|
||||
* @var \Mockery\MockInterface
|
||||
*/
|
||||
private $powerRepository;
|
||||
|
||||
/**
|
||||
* @var \Mockery\MockInterface
|
||||
*/
|
||||
private $initiateBackupService;
|
||||
|
||||
/**
|
||||
* @var \Mockery\MockInterface
|
||||
*/
|
||||
private $taskRepository;
|
||||
|
||||
/**
|
||||
* @var \Mockery\MockInterface
|
||||
*/
|
||||
private $scheduleRepository;
|
||||
|
||||
/**
|
||||
* Setup tests.
|
||||
*/
|
||||
public function setUp(): void
|
||||
{
|
||||
parent::setUp();
|
||||
|
||||
Bus::fake();
|
||||
Carbon::setTestNow(Carbon::now());
|
||||
|
||||
$this->commandRepository = m::mock(DaemonCommandRepository::class);
|
||||
$this->powerRepository = m::mock(DaemonPowerRepository::class);
|
||||
$this->taskRepository = m::mock(TaskRepository::class);
|
||||
$this->initiateBackupService = m::mock(InitiateBackupService::class);
|
||||
$this->scheduleRepository = m::mock(ScheduleRepository::class);
|
||||
|
||||
$this->app->instance(TaskRepositoryInterface::class, $this->taskRepository);
|
||||
$this->app->instance(ScheduleRepositoryInterface::class, $this->scheduleRepository);
|
||||
}
|
||||
|
||||
/**
|
||||
* Test power option passed to job.
|
||||
*/
|
||||
public function testPowerAction()
|
||||
{
|
||||
/** @var \Pterodactyl\Models\Schedule $schedule */
|
||||
$schedule = factory(Schedule::class)->make(['is_active' => true]);
|
||||
|
||||
/** @var \Pterodactyl\Models\Task $task */
|
||||
$task = factory(Task::class)->make(['action' => 'power', 'sequence_id' => 1]);
|
||||
|
||||
/* @var \Pterodactyl\Models\Server $server */
|
||||
$task->setRelation('server', $server = factory(Server::class)->make());
|
||||
$task->setRelation('schedule', $schedule);
|
||||
$server->setRelation('user', factory(User::class)->make());
|
||||
|
||||
$this->taskRepository->expects('getTaskForJobProcess')->with($task->id)->andReturn($task);
|
||||
$this->powerRepository->expects('setServer')->with($task->server)->andReturnSelf()
|
||||
->getMock()->expects('send')->with($task->payload)->andReturn(new Response);
|
||||
|
||||
$this->taskRepository->shouldReceive('update')->with($task->id, ['is_queued' => false])->once()->andReturnNull();
|
||||
$this->taskRepository->shouldReceive('getNextTask')->with($schedule->id, $task->sequence_id)->once()->andReturnNull();
|
||||
|
||||
$this->scheduleRepository->shouldReceive('withoutFreshModel->update')->with($schedule->id, [
|
||||
'is_processing' => false,
|
||||
'last_run_at' => Chronos::now()->toDateTimeString(),
|
||||
])->once()->andReturnNull();
|
||||
|
||||
$this->getJobInstance($task->id, $schedule->id);
|
||||
|
||||
Bus::assertNotDispatched(RunTaskJob::class);
|
||||
}
|
||||
|
||||
/**
|
||||
* Test command action passed to job.
|
||||
*/
|
||||
public function testCommandAction()
|
||||
{
|
||||
$schedule = factory(Schedule::class)->make();
|
||||
$task = factory(Task::class)->make(['action' => 'command', 'sequence_id' => 1]);
|
||||
$task->setRelation('server', $server = factory(Server::class)->make());
|
||||
$task->setRelation('schedule', $schedule);
|
||||
$server->setRelation('user', factory(User::class)->make());
|
||||
|
||||
$this->taskRepository->expects('getTaskForJobProcess')->with($task->id)->andReturn($task);
|
||||
$this->commandRepository->expects('setServer')->with($task->server)->andReturnSelf()
|
||||
->getMock()->expects('send')->with($task->payload)->andReturn(new Response);
|
||||
|
||||
$this->taskRepository->expects('update')->with($task->id, ['is_queued' => false])->andReturnNull();
|
||||
$this->taskRepository->expects('getNextTask')->with($schedule->id, $task->sequence_id)->andReturnNull();
|
||||
|
||||
$this->scheduleRepository->shouldReceive('withoutFreshModel->update')->with($schedule->id, [
|
||||
'is_processing' => false,
|
||||
'last_run_at' => Chronos::now()->toDateTimeString(),
|
||||
])->once()->andReturnNull();
|
||||
|
||||
$this->getJobInstance($task->id, $schedule->id);
|
||||
|
||||
Bus::assertNotDispatched(RunTaskJob::class);
|
||||
}
|
||||
|
||||
/**
|
||||
* Test that the next task in the list is queued if the current one is not the last.
|
||||
*/
|
||||
public function testNextTaskQueuedIfExists()
|
||||
{
|
||||
$schedule = factory(Schedule::class)->make();
|
||||
$task = factory(Task::class)->make(['action' => 'command', 'sequence_id' => 1]);
|
||||
$task->setRelation('server', $server = factory(Server::class)->make());
|
||||
$task->setRelation('schedule', $schedule);
|
||||
$server->setRelation('user', factory(User::class)->make());
|
||||
|
||||
$this->taskRepository->expects('getTaskForJobProcess')->with($task->id)->andReturn($task);
|
||||
$this->commandRepository->expects('setServer')->with($task->server)->andReturnSelf()
|
||||
->getMock()->expects('send')->with($task->payload)->andReturn(new Response);
|
||||
|
||||
$this->taskRepository->shouldReceive('update')->with($task->id, ['is_queued' => false])->once()->andReturnNull();
|
||||
|
||||
$nextTask = factory(Task::class)->make();
|
||||
$this->taskRepository->expects('getNextTask')->with($schedule->id, $task->sequence_id)->andReturn($nextTask);
|
||||
$this->taskRepository->expects('update')->with($nextTask->id, [
|
||||
'is_queued' => true,
|
||||
])->andReturnNull();
|
||||
|
||||
$this->getJobInstance($task->id, $schedule->id);
|
||||
|
||||
Bus::assertDispatched(RunTaskJob::class, function ($job) use ($nextTask, $schedule) {
|
||||
$this->assertEquals($nextTask->id, $job->task, 'Assert correct task ID is passed to job.');
|
||||
$this->assertEquals($schedule->id, $job->schedule, 'Assert correct schedule ID is passed to job.');
|
||||
$this->assertEquals($nextTask->time_offset, $job->delay, 'Assert correct job delay time is set.');
|
||||
|
||||
return true;
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Test that an exception is thrown if an invalid task action is supplied.
|
||||
*/
|
||||
public function testInvalidActionPassedToJob()
|
||||
{
|
||||
$this->expectException(InvalidArgumentException::class);
|
||||
$this->expectExceptionMessage('Cannot run a task that points to a non-existent action.');
|
||||
|
||||
$schedule = factory(Schedule::class)->make();
|
||||
$task = factory(Task::class)->make(['action' => 'invalid', 'sequence_id' => 1]);
|
||||
$task->setRelation('server', $server = factory(Server::class)->make());
|
||||
$task->setRelation('schedule', $schedule);
|
||||
$server->setRelation('user', factory(User::class)->make());
|
||||
|
||||
$this->taskRepository->expects('getTaskForJobProcess')->with($task->id)->andReturn($task);
|
||||
|
||||
$this->getJobInstance($task->id, 1234);
|
||||
}
|
||||
|
||||
/**
|
||||
* Test that a schedule marked as disabled does not get processed.
|
||||
*/
|
||||
public function testScheduleMarkedAsDisabledDoesNotProcess()
|
||||
{
|
||||
$schedule = factory(Schedule::class)->make(['is_active' => false]);
|
||||
$task = factory(Task::class)->make(['action' => 'invalid', 'sequence_id' => 1]);
|
||||
$task->setRelation('server', $server = factory(Server::class)->make());
|
||||
$task->setRelation('schedule', $schedule);
|
||||
$server->setRelation('user', factory(User::class)->make());
|
||||
|
||||
$this->taskRepository->shouldReceive('getTaskForJobProcess')->with($task->id)->once()->andReturn($task);
|
||||
|
||||
$this->scheduleRepository->shouldReceive('withoutFreshModel->update')->with($schedule->id, [
|
||||
'is_processing' => false,
|
||||
'last_run_at' => Chronos::now()->toDateTimeString(),
|
||||
])->once()->andReturn(1);
|
||||
|
||||
$this->taskRepository->shouldReceive('update')->with($task->id, ['is_queued' => false])->once()->andReturn(1);
|
||||
|
||||
$this->getJobInstance($task->id, $schedule->id);
|
||||
$this->assertTrue(true);
|
||||
}
|
||||
|
||||
/**
|
||||
* Run the job using the mocks provided.
|
||||
*
|
||||
* @param int $task
|
||||
* @param int $schedule
|
||||
*
|
||||
* @throws \Pterodactyl\Exceptions\Model\DataValidationException
|
||||
*/
|
||||
private function getJobInstance($task, $schedule)
|
||||
{
|
||||
return (new RunTaskJob($task, $schedule))->handle(
|
||||
$this->commandRepository,
|
||||
$this->initiateBackupService,
|
||||
$this->powerRepository,
|
||||
$this->taskRepository
|
||||
);
|
||||
}
|
||||
}
|
Loading…
Reference in a new issue