24634959708

Committed 19 Apr 2026 05:31PM UTC coverage: 95.05% (-2.8%) from 97.882%

Build # 24634959708

Build Type

push

github

Committed by

mabar

Commit Message

Move events from subprocesses to main process, fire after job callbacks for locked jobs and maintenance, parse subprocess events via new protocol

Coverage Stats

171 of 223 new or added lines in 4 files covered. (76.68%)

54 existing lines in 3 files now uncovered.

3053 of 3212 relevant lines covered (95.05%)

76.83 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

61.63

/src/Command/RunJobCommand.php

<?php declare(strict_types = 1);

namespace Orisai\Scheduler\Command;

use Closure;
use Orisai\Clock\SystemClock;
use Orisai\Scheduler\Exception\JobFailure;
use Orisai\Scheduler\Executor\SubprocessEventProtocol;
use Orisai\Scheduler\Scheduler;
use Orisai\Scheduler\Status\JobInfo;
use Orisai\Scheduler\Status\JobResult;
use Orisai\Scheduler\Status\JobResultState;
use Orisai\Scheduler\Status\JobSummary;
use Orisai\Scheduler\Status\RunParameters;
use Psr\Clock\ClockInterface;
use Symfony\Component\Console\Input\InputArgument;
use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Input\InputOption;
use Symfony\Component\Console\Output\OutputInterface;
use Throwable;
use function fflush;
use function fwrite;
use function get_class;
use function json_decode;
use function json_encode;
use function ob_end_clean;
use function ob_get_clean;
use function ob_get_contents;
use function ob_start;
use const JSON_PRETTY_PRINT;
use const JSON_THROW_ON_ERROR;
use const STDOUT;

final class RunJobCommand extends BaseRunCommand
{

        private Scheduler $scheduler;

        public function __construct(Scheduler $scheduler, ?ClockInterface $clock = null)
        {
                parent::__construct($clock ?? new SystemClock());
                $this->scheduler = $scheduler;
        }

        public static function getDefaultName(): string
        {
                return 'scheduler:run-job';
        }

        public static function getDefaultDescription(): string
        {
                return 'Run single job, ignoring scheduled time';
        }

        protected function configure(): void
        {
                $this->addArgument('id', InputArgument::REQUIRED, 'Job ID (visible in scheduler:list)');
                $this->addOption(
                        'no-force',
                        null,
                        InputOption::VALUE_NONE,
                        'Don\'t force job to run and respect due time instead',
                );
                $this->addOption('json', null, InputOption::VALUE_NONE, 'Output in json format');
                $this->addOption('parameters', null, InputOption::VALUE_REQUIRED, '[Internal]');
                $this->addOption('events', null, InputOption::VALUE_NONE, '[Internal] Emit framework events on stdout');
        }

        protected function execute(InputInterface $input, OutputInterface $output): int
        {
                $json = $input->getOption('json');
                $params = $input->getOption('parameters');
                $events = $input->getOption('events');

                [$onStarted, $onFinished] = $events
                        ? $this->createEventEmitters()
                        : [null, null];

                ob_start(static fn () => null);
                try {
                        $summary = $this->scheduler->runJob(
                                $input->getArgument('id'),
                                !$input->getOption('no-force'),
                                $params === null
                                        ? null
                                        : RunParameters::fromArray(json_decode($params, true, 512, JSON_THROW_ON_ERROR)),
                                $onStarted,
                                $onFinished,
                        );

                        $stdout = ($tmp = ob_get_clean()) === false ? '' : $tmp;
                } catch (JobFailure $e) {
                        ob_end_clean();

                        // In --events mode, the subprocess already emitted the `finished` event via the
                        // afterJobEmitter inside runInternal. Emit a `failure` event so the parent knows
                        // the job's throwable was not handled by an errorHandler and can propagate a
                        // RunFailure — then exit cleanly so the parent treats the subprocess as succeeded
                        // at reporting (rather than a crashed subprocess).
                        if ($events) {
                                $previous = $e->getPrevious();
                                self::emitEvent([
                                        'type' => SubprocessEventProtocol::TypeFailure,
                                        'class' => get_class($previous),
                                        'message' => $previous->getMessage(),
                                ]);

                                return $this->exitCodeFromState($e->getSummary());
                        }

                        throw $e;
                } catch (Throwable $e) {
                        ob_end_clean();

                        throw $e;
                }

                if ($events) {
                        // `finished` event already emitted via emitter — no additional stdout needed.
                        return $summary === null ? self::SUCCESS : $this->exitCodeFromState($summary);
                }

                if ($summary === null) {
                        if ($json) {
                                $output->writeln(json_encode(null, JSON_THROW_ON_ERROR));
                        } else {
                                $output->writeln('<info>Command was not executed because it is not its due time</info>');
                        }

                        return self::SUCCESS;
                }

                if ($json) {
                        $this->renderJobAsJson($summary, $stdout, $output);
                } else {
                        if ($stdout !== '') {
                                $output->writeln($stdout);
                        }

                        $this->renderJob($summary, $this->getTerminalWidth(), $output);
                }

                return $this->exitCodeFromState($summary);
        }

        /**
         * @return array{0: Closure(JobInfo): void, 1: Closure(JobInfo, JobResult): void}
         */
        private function createEventEmitters(): array
        {
                $onStarted = static function (JobInfo $info): void {
                        self::emitEvent([
                                'type' => SubprocessEventProtocol::TypeStarted,
                                'info' => $info->toArray(),
                        ]);
                };

                $onFinished = static function (JobInfo $info, JobResult $result): void {
                        $captured = ob_get_contents();
                        self::emitEvent([
                                'type' => SubprocessEventProtocol::TypeFinished,
                                'info' => $info->toArray(),
                                'result' => $result->toArray(),
                                'stdout' => $captured === false ? '' : $captured,
                        ]);
                };

                return [$onStarted, $onFinished];
        }

        /**
         * @param array<mixed> $payload
         */
        private static function emitEvent(array $payload): void
        {
                $line = SubprocessEventProtocol::EventMarker
                        . json_encode($payload, JSON_THROW_ON_ERROR)
                        . "\n";

                fwrite(STDOUT, $line);
                fflush(STDOUT);
        }

        private function exitCodeFromState(JobSummary $summary): int
        {
                return $summary->getResult()->getState() === JobResultState::fail()
                        ? self::FAILURE
                        : self::SUCCESS;
        }

        private function renderJobAsJson(JobSummary $summary, string $stdout, OutputInterface $output): void
        {
                $jobData = $summary->toArray() + ['stdout' => $stdout];

                $output->writeln(
                        json_encode($jobData, JSON_THROW_ON_ERROR | JSON_PRETTY_PRINT),
                );
        }

}

1	<?php declare(strict_types = 1);
2
3	namespace Orisai\Scheduler\Command;
4
5	use Closure;
6	use Orisai\Clock\SystemClock;
7	use Orisai\Scheduler\Exception\JobFailure;
8	use Orisai\Scheduler\Executor\SubprocessEventProtocol;
9	use Orisai\Scheduler\Scheduler;
10	use Orisai\Scheduler\Status\JobInfo;
11	use Orisai\Scheduler\Status\JobResult;
12	use Orisai\Scheduler\Status\JobResultState;
13	use Orisai\Scheduler\Status\JobSummary;
14	use Orisai\Scheduler\Status\RunParameters;
15	use Psr\Clock\ClockInterface;
16	use Symfony\Component\Console\Input\InputArgument;
17	use Symfony\Component\Console\Input\InputInterface;
18	use Symfony\Component\Console\Input\InputOption;
19	use Symfony\Component\Console\Output\OutputInterface;
20	use Throwable;
21	use function fflush;
22	use function fwrite;
23	use function get_class;
24	use function json_decode;
25	use function json_encode;
26	use function ob_end_clean;
27	use function ob_get_clean;
28	use function ob_get_contents;
29	use function ob_start;
30	use const JSON_PRETTY_PRINT;
31	use const JSON_THROW_ON_ERROR;
32	use const STDOUT;
33
34	final class RunJobCommand extends BaseRunCommand
35	{
36
37	private Scheduler $scheduler;
38
39	public function __construct(Scheduler $scheduler, ?ClockInterface $clock = null)
40	{
41	parent::__construct($clock ?? new SystemClock());	56✔
42	$this->scheduler = $scheduler;	56✔
43	}
44
45	public static function getDefaultName(): string
46	{
47	return 'scheduler:run-job';	56✔
48	}
49
50	public static function getDefaultDescription(): string
51	{
52	return 'Run single job, ignoring scheduled time';	56✔
53	}
54
55	protected function configure(): void
56	{
57	$this->addArgument('id', InputArgument::REQUIRED, 'Job ID (visible in scheduler:list)');	56✔
58	$this->addOption(	56✔
59	'no-force',	56✔
60	null,	56✔
61	InputOption::VALUE_NONE,	56✔
62	'Don\'t force job to run and respect due time instead',	56✔
63	);	56✔
64	$this->addOption('json', null, InputOption::VALUE_NONE, 'Output in json format');	56✔
65	$this->addOption('parameters', null, InputOption::VALUE_REQUIRED, '[Internal]');	56✔
66	$this->addOption('events', null, InputOption::VALUE_NONE, '[Internal] Emit framework events on stdout');	56✔
67	}
68
69	protected function execute(InputInterface $input, OutputInterface $output): int
70	{
71	$json = $input->getOption('json');	56✔
72	$params = $input->getOption('parameters');	56✔
73	$events = $input->getOption('events');	56✔
74
75	[$onStarted, $onFinished] = $events	56✔
NEW 76	? $this->createEventEmitters()	×
77	: [null, null];	56✔
78
79	ob_start(static fn () => null);	56✔
80	try {
81	$summary = $this->scheduler->runJob(	56✔
82	$input->getArgument('id'),	56✔
83	!$input->getOption('no-force'),	56✔
84	$params === null	56✔
85	? null	56✔
86	: RunParameters::fromArray(json_decode($params, true, 512, JSON_THROW_ON_ERROR)),	56✔
87	$onStarted,	56✔
88	$onFinished,	56✔
89	);	56✔
90
91	$stdout = ($tmp = ob_get_clean()) === false ? '' : $tmp;	48✔
92	} catch (JobFailure $e) {	8✔
NEW 93	ob_end_clean();	×
94
95	// In --events mode, the subprocess already emitted the `finished` event via the
96	// afterJobEmitter inside runInternal. Emit a `failure` event so the parent knows
97	// the job's throwable was not handled by an errorHandler and can propagate a
98	// RunFailure — then exit cleanly so the parent treats the subprocess as succeeded
99	// at reporting (rather than a crashed subprocess).
NEW 100	if ($events) {	×
NEW 101	$previous = $e->getPrevious();	×
NEW 102	self::emitEvent([	×
NEW 103	'type' => SubprocessEventProtocol::TypeFailure,	×
NEW 104	'class' => get_class($previous),	×
NEW 105	'message' => $previous->getMessage(),	×
NEW 106	]);	×
107
NEW 108	return $this->exitCodeFromState($e->getSummary());	×
109	}
110
NEW 111	throw $e;	×
112	} catch (Throwable $e) {	8✔
113	ob_end_clean();	8✔
114
115	throw $e;	8✔
116	}
117
118	if ($events) {	48✔
119	// `finished` event already emitted via emitter — no additional stdout needed.
NEW 120	return $summary === null ? self::SUCCESS : $this->exitCodeFromState($summary);	×
121	}
122
123	if ($summary === null) {	48✔
124	if ($json) {	16✔
125	$output->writeln(json_encode(null, JSON_THROW_ON_ERROR));	8✔
126	} else {
127	$output->writeln('<info>Command was not executed because it is not its due time</info>');	8✔
128	}
129
130	return self::SUCCESS;	16✔
131	}
132
133	if ($json) {	48✔
134	$this->renderJobAsJson($summary, $stdout, $output);	16✔
135	} else {
136	if ($stdout !== '') {	40✔
137	$output->writeln($stdout);	8✔
138	}
139
140	$this->renderJob($summary, $this->getTerminalWidth(), $output);	40✔
141	}
142
143	return $this->exitCodeFromState($summary);	48✔
144	}
145
146	/**
147	* @return array{0: Closure(JobInfo): void, 1: Closure(JobInfo, JobResult): void}
148	*/
149	private function createEventEmitters(): array
150	{
NEW 151	$onStarted = static function (JobInfo $info): void {	×
NEW 152	self::emitEvent([	×
NEW 153	'type' => SubprocessEventProtocol::TypeStarted,	×
NEW 154	'info' => $info->toArray(),	×
NEW 155	]);	×
NEW 156	};	×
157
NEW 158	$onFinished = static function (JobInfo $info, JobResult $result): void {	×
NEW 159	$captured = ob_get_contents();	×
NEW 160	self::emitEvent([	×
NEW 161	'type' => SubprocessEventProtocol::TypeFinished,	×
NEW 162	'info' => $info->toArray(),	×
NEW 163	'result' => $result->toArray(),	×
NEW 164	'stdout' => $captured === false ? '' : $captured,	×
NEW 165	]);	×
NEW 166	};	×
167
NEW 168	return [$onStarted, $onFinished];	×
169	}
170
171	/**
172	* @param array<mixed> $payload
173	*/
174	private static function emitEvent(array $payload): void
175	{
NEW 176	$line = SubprocessEventProtocol::EventMarker	×
NEW 177	. json_encode($payload, JSON_THROW_ON_ERROR)	×
NEW 178	. "\n";	×
179
NEW 180	fwrite(STDOUT, $line);	×
NEW 181	fflush(STDOUT);	×
182	}
183
184	private function exitCodeFromState(JobSummary $summary): int
185	{
186	return $summary->getResult()->getState() === JobResultState::fail()	48✔
187	? self::FAILURE	8✔
188	: self::SUCCESS;	48✔
189	}
190
191	private function renderJobAsJson(JobSummary $summary, string $stdout, OutputInterface $output): void
192	{
193	$jobData = $summary->toArray() + ['stdout' => $stdout];	16✔
194
195	$output->writeln(	16✔
196	json_encode($jobData, JSON_THROW_ON_ERROR \| JSON_PRETTY_PRINT),	16✔
197	);	16✔
198	}
199
200	}

orisai / scheduler / 24634959708

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous