PHPackages bylexus/php-tr - PHPackages - PHPackages [Skip to content](#main-content)[PHPackages](/)[Directory](/)[Categories](/categories)[Trending](/trending)[Leaderboard](/leaderboard)[Changelog](/changelog)[Analyze](/analyze)[Collections](/collections)[Log in](/login)[Sign up](/register) 1. [Directory](/) 2. / 3. [Queues & Workers](/categories/queues) 4. / 5. bylexus/php-tr ActiveLibrary[Queues & Workers](/categories/queues) bylexus/php-tr ============== Queued Task worker library for PHP backed by a relational database 0.5.0(1mo ago)0119MITPHPPHP ^8.3CI passing Since May 15Pushed 1mo agoCompare [ Source](https://github.com/bylexus/php-tr)[ Packagist](https://packagist.org/packages/bylexus/php-tr)[ RSS](/packages/bylexus-php-tr/feed)WikiDiscussions main Synced 3w ago READMEChangelog (5)Dependencies (10)Versions (8)Used By (0) PHP Task Runner - a Queue / Task Runner for background tasks ============================================================ [](#php-task-runner---a-queue--task-runner-for-background-tasks) [![Build Test](https://github.com/bylexus/php-tr/actions/workflows/build-test.yml/badge.svg)](https://github.com/bylexus/php-tr/actions/workflows/build-test.yml) ⚠️ **Work in progress! Use with caution for now! Very early version!** ⚠️ PHP Task Runner is a database-backed workflow library for PHP >= 8.3. It is meant to queue and run jobs that are to be processed in the background of a frontend application (e.g. queue an email to be sent in the background). It is not a standalone queue/runner, but meant to be integrated in your / a existing framework. So this library needs to be integrated in your environment. You model work as a `Task` that defines a workflow consisting of `Step`s. The Task defines the order of the Steps to be processed. Enqueued Tasks then get worked on step-by-step by a Runner. The library stores the Task state in the database so multiple runner processes can safely compete for queued work. The public surface is intentionally small and framework agnostic: - `Task` defines the workflow, owns the payload, and carries all queue state (status, step class, retry attempt, timestamps, ...). - `Step` is an interface implemented by classes that contain the actual work in their `execute()` method. Step instances are stateless — every piece of persisted state lives on the Task. - `Runner` claims queued Tasks, instantiates the configured Step, executes it, and persists the next state. `Task` and `Step` classes are usually kept separately so that single-purpose `Step` classes can be mixed and matched by several `Task` classes. For example, a generic `SendMail` step can be used by many tasks to send information emails. For trivial single-step tasks, a single class can implement both `Task` and `Step`. This README is written for experienced PHP developers who want to integrate the library into an existing application or framework. Requirements ------------ [](#requirements) - PHP 8.3+ - `ext-pdo` - One of the supported PDO backends: - PostgreSQL - MySQL >= 8.0 - MariaDB >= 10.6 - SQLite - Autoloadable task and step classes in every process that enqueues or runs work Installation ------------ [](#installation) ``` composer require bylexus/php-tr ``` Quickstart ---------- [](#quickstart) See a full example in [examples/quickstart.php](examples/quickstart.php). This quickstart just implements a single Task with a single Step to work on. This chapter explains how to get startet in detail. ### Setup [](#setup) Create a `TaskEnvironment` instance: the `TaskEnvironment` is the configuration object that contains all the needed dependencies: ``` use ByLexus\TaskRunner\TaskEnvironment; use ByLexus\TaskRunner\Queue\QueueConfiguration; $env = new TaskEnvironment( connection: getDBConn(), // get your PDO connection as usual queueConfiguration: new QueueConfiguration(schemaName: 'appschema'), // ... ); ``` The env object will be used for all interaction with the Tasks / runner. Create the DB objects automatically: ``` // Reuse the same TaskEnvironment for schema management, enqueueing, and runners. $env->getSchemaManager()->bootstrap(); ``` or by exporting the needed SQL through your own tooling: ``` $ddl = $env->getSchemaManager()->exportDdl(); // Dump, log, or feed $ddl into your migration tooling. ``` ### Create a Step class [](#create-a-step-class) First you define one (or multiple) single-purpose Step classes. Steps are one piece of work that can be used in a / multiple Task. A Step class implements the `Step` interface (a single `execute()` method). Here, we create a simple Step that just prints a message: ``` use ByLexus\TaskRunner\Step; use ByLexus\TaskRunner\Task; use ByLexus\TaskRunner\Result\StepResult; final class PrintGreetingStep implements Step { // Implement the execute function to execute the work: public function execute(Task $task): StepResult { // Steps read input from the task payload. // It is advisable to use a namespaced payload, as all steps of a task share // the same Payload object. Here, we use the class name as namespace: $name = self::recipient($task); // Do the work! fwrite(STDOUT, sprintf("Hello %s from a step.\n", $recipient)); // and return a result: return StepResult::succeeded(message: 'Greeting printed.'); } // Helper functions to get/set values from the Task's payload: public static function setName(Task $task, string $recipient) { $task->getPayload(static::class)->recipient = $recipient; } public static function recipient(Task $task): string { return $task->getPayload(static::class)->recipient ?? 'world'; } } ``` Step instances are stateless: all persistent workflow state (current step class, retry attempt, started/finished timestamps, payload, ...) lives on the Task. Inside `execute()`, read step lifecycle data via `$task->getStepAttempt()`, `$task->getStepStartedAt()`, etc. ### Create a Task class to define your workflow [](#create-a-task-class-to-define-your-workflow) Now, define the `Task` class to define your workflow: Define the needed payload data used by your steps, and create a workflow in the `nextStep` function: ``` use ByLexus\TaskRunner\Step; use ByLexus\TaskRunner\Task; use ByLexus\TaskRunner\Attribute\CleanupAfter; #[CleanupAfter(new DateInterval('PT10M'), new DateInterval('P7D'))] final class GreetingTask extends Task { // withRecipient is just a helper method to set up the correct payload: public function withRecipient(string $recipient): self { // The root payload is just a stdClass, so examples can keep setup lightweight. $this->getPayload()->globalValue = 'some global value'; // You need to know the exact payload path for providing data for later steps: // Here, we use the static function defined in the step PrintGreeting: PrintGreetingStep::setRecipient($this, $recipient); return $this; } // nextStep allows the Task to form a workflow: // it receives the actual (done) step and can now return the next (configured) step. // Returning null means the flow is done: public function nextStep(?Step $actStep = null): ?Step { // Returning null ends the workflow. Returning a step queues the next unit of work. return $actStep === null ? new PrintGreetingStep() : null; } } ``` ### Dispatch a task [](#dispatch-a-task) Now you're ready to dispatch the task: ``` // The task owns the payload. Here we seed it before enqueueing the first step. $task = (new GreetingTask())->withRecipient('Ada Lovelace'); $env->enqueue($task); // Optional: lower numbers are picked first by runners. $env->enqueue($task, priority: Task::PRIO_HIGH); ``` Tasks default to priority `3` (`Task::PRIO_NORMAL`). You can choose values from `1` to `5` using the built-in constants: - `Task::PRIO_VERY_HIGH` = `1` - `Task::PRIO_HIGH` = `2` - `Task::PRIO_NORMAL` = `3` - `Task::PRIO_LOW` = `4` - `Task::PRIO_VERY_LOW` = `5` When multiple queued tasks are available, runners claim the highest-priority work first, then fall back to `available_at` and creation time. If your queued tasks need constructor injection, configure the container and logger once on `TaskEnvironment` and reuse that same object for enqueueing, lookup, and runner creation. ### Start a runner to work on the tasks [](#start-a-runner-to-work-on-the-tasks) A Runner can now be instantiated in a separate script, e.g. a script that runs server-side as a daemon: ``` use ByLexus\TaskRunner\Queue\QueueConfiguration; use ByLexus\TaskRunner\TaskEnvironment; use ByLexus\TaskRunner\RunnerConfiguration; use Psr\Log\LoggerInterface; $queueConfiguration = new QueueConfiguration('app_background_jobs', 'background_jobs'); $env = new TaskEnvironment( connection: $pdo, queueConfiguration: $queueConfiguration, container: $container, // provide your PSR-11 compatible Dependency Injection container logger: $container->get(LoggerInterface::class), // PSR-3 compatible logger runnerConfiguration: new RunnerConfiguration(runnerId: 'app-worker-1'), ); // A runner claims one queued row, hydrates the task and step, executes them, and persists the result. $runner = $env->createRunner(); $runner->runLoop(); ``` ⚠️ **Note for stopping the runner**: ⚠️ On systems that support signal handling with the PHP `pcntl` module, a safe shutdown is initiated on `SIGINT` and `SIGTERM` (aka CTRL-C and soft kill). Make sure to install / enable the PHP extension `pcntl` to support signal handling. Systems that do NOT support posix / pcntl signal handling just kill the runner - even if it is working on Tasks. Concepts -------- [](#concepts) ### `Task` is the workflow instance [](#task-is-the-workflow-instance) A task is the long-lived object stored in the queue row. It owns the payload (arbitary data for the steps) and decides the workflow graph through `nextStep()`. ``` abstract class Task { abstract public function nextStep(?Step $actStep = null): ?Step; } ``` Important points: - The task is the only payload owner. Steps receive the task and read or mutate payload through `$task->getPayload()`. - The steps itself read the payload data, so you have to know the exact name / path of the payload. - The runner persists the task payload after every step execution. - The runner calls `nextStep()` of your task to fetch the next work unit. Return `null` to indicate done. ### `Step` is one unit of work [](#step-is-one-unit-of-work) Every step implements `execute(Task $task): StepResult`. ``` interface Step { public function execute(Task $task): StepResult; } ``` Use a step for one unit of work. A step should be undependant of other steps / tasks: It receives its information from the Payload of the task (and can also modify it). Step classes are instantiated fresh on each run and carry no persisted state. If your step needs collaborators (mailer, HTTP client, logger, ...), declare them as constructor parameters: the runner resolves them through the configured PSR-11 container. ### Single-class Task + Step [](#single-class-task--step) For trivial workflows with exactly one step, a single class can implement both the `Task` abstract class and the `Step` interface. The runner detects when `task_class` equals `step_class` and reuses the same instance: ``` use ByLexus\TaskRunner\Result\StepResult; use ByLexus\TaskRunner\Step; use ByLexus\TaskRunner\Task; final class SendReminderTask extends Task implements Step { public function withEmail(string $email): self { $this->getPayload()->email = $email; return $this; } public function nextStep(?Step $actStep = null): ?Step { return $actStep === null ? $this : null; } public function execute(Task $task): StepResult { // send reminder to $task->getPayload()->email return StepResult::succeeded(); } } ``` After the work is done (or failed), the step must return a `StepResult` indicating the state of the result. The result itself must not contain data for futher steps (that goes to the payload), but result information only. - return `StepResult::succeeded()` when the step was successful - return `StepResult::failed()` with `ErrorInfo` when the step failed - throw an exception when it cannot recover locally; the runner converts that exception into a failed `StepResult` ### `Runner` is the worker process [](#runner-is-the-worker-process) The runner claims one queued task/step at a time, hydrates the task and step class from the row, executes the step, and persists one of these outcomes: - queue the same step again for a retry - queue the next step - mark the task as `succeeded`, `failed` or `cancelled` Use: - `runSingle()` for cron-style polling, tests, or one-shot commands - `runLoop()` for a long-running worker process, waiting for new tasks `runLoop()` uses PostgreSQL `LISTEN` / `NOTIFY` only when the active PDO connection supports it. MySQL, MariaDB, and SQLite use the polling variant only and sleep for the configured wait timeout between claim attempts when no task is available. You can safely start multiple runners, as each task can only be claimed by one runner at a time: This allows for parallel execution of multiple tasks. Useful if your runner gets blocked with long-running tasks. ### Task priority [](#task-priority) Each queued task row stores a numeric priority. Priority `1` is the highest priority and `5` is the lowest. If you do not pass a priority when enqueueing, the library stores `3`. ``` use ByLexus\TaskRunner\TaskEnvironment; use ByLexus\TaskRunner\Task; $env = new TaskEnvironment($pdo); $task = (new WelcomeTask())->withEmail('ada@example.com'); $env->enqueue($task, priority: Task::PRIO_VERY_HIGH); ``` This is useful when some background work should jump ahead of normal queue traffic without needing a separate queue table. Defining Tasks And Steps ------------------------ [](#defining-tasks-and-steps) This is the smallest useful pattern: ```