*/ private array $handlers = []; public function __construct(private Queue $queue, private string $workerId) { } /** @param callable(Job): void $handler */ public function handle(string $type, callable $handler): void { $this->handlers[$type] = $handler; } /** * @param list|null $onlyTypes when given, only jobs of these types are claimed (a manual run never touches other kinds of work) * @param int|null $maxSeconds when given, no new job is started once this many seconds have passed (a job already running finishes) * @return array{ran:int, failed:int} */ public function runBatch(int $maxJobs = 10, int $leaseSeconds = 120, ?array $onlyTypes = null, ?int $maxSeconds = null): array { $ran = 0; $failed = 0; $types = array_keys($this->handlers); if ($onlyTypes !== null) { $types = array_values(array_intersect($types, $onlyTypes)); } if ($types === []) { return ['ran' => 0, 'failed' => 0]; } $started = microtime(true); for ($i = 0; $i < $maxJobs; $i++) { if ($maxSeconds !== null && $i > 0 && microtime(true) - $started >= $maxSeconds) { break; } $job = $this->queue->claim($this->workerId, $leaseSeconds, $types); if ($job === null) { break; } try { ($this->handlers[$job->type])($job); $this->queue->complete($job, $this->workerId); $ran++; } catch (PermanentJobFailure $e) { $this->queue->fail($job, $this->workerId, $e->getMessage(), false); $failed++; } catch (\Throwable $e) { $this->queue->fail($job, $this->workerId, $e::class . ': ' . $e->getMessage(), true); $failed++; } } return ['ran' => $ran, 'failed' => $failed]; } }