PHP API

Просмотреть как Markdown Редактировать на GitHub Документация по 2cc7b2b

Всё в Queuety доступно через статический класс Queuety\Queuety. Это единая точка входа для отправки задач, создания рабочих процессов, управления конечными автоматами и очередями, а также доступа к внутренним подсистемам.

PHP

use Queuety\Queuety;

Доступ к автономному ядру

Среду выполнения Queuety с поддержкой БД можно инициализировать вне WordPress.

Этот автономный режим предназначен для ядра оркестрации:

задачи и рабочие процессы (для пулов рабочих процессов также нужны константы DB_HOST, DB_NAME, DB_USER и DB_PASSWORD; см. ниже)
рабочие процессы и сигналы
конечные автоматы и события
артефакты, журналы, метрики и расписания

Это не замена адаптеру WordPress. Queuety::on_action(), replace_wp_cron(), загрузка плагинов и реальная регистрация wp queuety ... по-прежнему требуют WordPress.

`Queuety::init( Connection $conn )`

Инициализируйте Queuety с прямым подключением к БД.

PHP

use Queuety\Connection;
use Queuety\Queuety;

$conn = new Connection(
    host: '127.0.0.1:3306',
    dbname: 'wordpress',
    user: 'root',
    password: 'secret',
    prefix: 'wp_',
);

Queuety::init( $conn );

Необязательный аргумент table_prefix изменяет часть Queuety в именах таблиц среды выполнения для каждого подключения:

PHP

$conn = new Connection(
    host: '127.0.0.1:3306',
    dbname: 'wordpress',
    user: 'root',
    password: 'secret',
    prefix: 'wp_',
    table_prefix: 'themequeue_',
);

Например, таблица queuety_jobs преобразуется в wp_themequeue_jobs.

`Queuety::ensure_schema()`

Убедитесь, что для активного подключения существуют таблицы среды выполнения Queuety.

PHP

Queuety::init( $conn );
Queuety::ensure_schema();

Описание полной модели и границ WordPress приведено в разделе Автономное использование.

Задачи

`Queuety::dispatch( $handler, $payload )`

Отправляет задачу на выполнение. Принимает строковое имя обработчика или экземпляр Contracts\Job. Возвращает объект построителя PendingJob для последовательной настройки параметров.

PHP

// Handler name + payload (original API)
Queuety::dispatch( 'send_email', [ 'to' => 'user@example.com' ] );

// Contracts\Job instance (v0.6.0+)
Queuety::dispatch( new SendEmailJob( to: 'user@example.com' ) );

// With options
Queuety::dispatch( 'send_email', [ 'to' => 'user@example.com' ] )
    ->on_queue( 'emails' )
    ->with_priority( Priority::High )
    ->delay( 300 )
    ->max_attempts( 5 )
    ->rate_limit( 10, 60 )
    ->unique()
    ->after( $other_job_id );

// Get the job ID
$job_id = Queuety::dispatch( 'send_email', $payload )->id();

Если передан экземпляр Contracts\Job, сериализатор извлекает полное квалифицированное имя класса (FQCN) в качестве имени обработчика, а открытые свойства — в качестве данных. Параметр $payload игнорируется, если $handler является экземпляром Job.

Методы `PendingJob`

Метод	Описание
`on_queue( string $queue )`	Задать целевую очередь
`with_priority( Priority $priority )`	Задать уровень приоритета
`delay( int $seconds )`	Задержать появление задачи в очереди
`max_attempts( int $max )`	Максимальное число попыток повтора
`rate_limit( int $max, int $window )`	Ограничение частоты: максимум выполнений за интервал в секундах
`unique()`	Предотвратить повторную отправку одной и той же задачи
`after( int $job_id )`	Дождаться завершения другой задачи
`concurrency_group( string $group, int $limit )`	Добавить задачу в общую группу допуска для рабочих процессов
`cost_units( int $units )`	Привязать к отправке метаданные об относительной стоимости выполнения
`id()`	Принудительно отправить задачу и вернуть её идентификатор

concurrency_group() и cost_units() входят в более широкую модель Управление ресурсами, поэтому одна отправка задачи может участвовать в ограничениях параллелизма по количеству, взвешенных бюджетах очередей или поставщиков и эвристиках допуска рабочих процессов.

`Queuety::dispatch_sync( $handler, $payload )`

Выполнить задачу синхронно в текущем процессе, не добавляя её в очередь.

PHP

// With a Job instance
Queuety::dispatch_sync( new SendEmailJob( to: 'user@example.com', subject: 'Hi', body: 'Hello' ) );

// With a handler name and payload
Queuety::dispatch_sync( 'send_email', [ 'to' => 'user@example.com' ] );

Метод handle() задачи вызывается напрямую. Строка в базе данных не создаётся, рабочий процесс не запускается, промежуточные обработчики не применяются.

`Queuety::batch( $jobs )`

Отправить несколько задач одним многострочным запросом INSERT.

PHP

$job_ids = Queuety::batch( [
    [ 'handler' => 'send_email', 'payload' => [ 'to' => 'a@example.com' ], 'queue' => 'emails' ],
    [ 'handler' => 'send_email', 'payload' => [ 'to' => 'b@example.com' ], 'priority' => 2 ],
] );

`Queuety::create_batch( $jobs )`

Создаёт BatchBuilder для отправки группы задач с обратными вызовами и отслеживанием выполнения. Подробности см. в разделе Объединение в группы.

PHP

$batch = Queuety::create_batch( [
    new ImportUsersJob( $chunk_1 ),
    new ImportUsersJob( $chunk_2 ),
] )
    ->name( 'Import users' )
    ->then( ImportCompleteHandler::class )
    ->catch( ImportFailedHandler::class )
    ->finally( ImportCleanupHandler::class )
    ->allow_failures()
    ->on_queue( 'imports' )
    ->dispatch();

Методы `BatchBuilder`

Метод	Возвращает	Описание
`name( string $name )`	`self`	Задать имя группы
`on_queue( string $queue )`	`self`	Задать очередь для всех задач в группе
`then( string $handler_class )`	`self`	Обработчик, вызываемый при успешном завершении группы
`catch( string $handler_class )`	`self`	Обработчик, вызываемый при сбое первой задачи в группе
`finally( string $handler_class )`	`self`	Обработчик, вызываемый при завершении группы (успешном или неудачном)
`allow_failures()`	`self`	Разрешить вызов обратного вызова `then`, даже если произошли сбои
`dispatch()`	`Batch`	Создать запись группы и все задачи; возвращает объект-значение `Batch`

`Queuety::find_batch( $id )`

Найти группу по идентификатору. Возвращает объект-значение Batch или null.

PHP

$batch = Queuety::find_batch( $batch_id );

if ( null !== $batch ) {
    echo $batch->progress() . '% complete';
}

Управление рабочими процессами

`Queuety::run_worker( $queue = 'default', $once = false )`

Запустить один цикл рабочего процесса напрямую.

PHP

Queuety::run_worker( 'default' );
Queuety::run_worker( 'critical,default', true );

`Queuety::run_worker_pool( $workers, $queue = 'default' )`

Напрямую запустить пул рабочих процессов фиксированного размера. Требуются расширения pcntl и posix. Для дочерних подключений пул считывает константы WordPress DB_HOST, DB_NAME, DB_USER и DB_PASSWORD (из инициализированного подключения берутся только префиксы таблиц), поэтому задайте их при запуске пула вне WordPress (#5).

PHP

Queuety::run_worker_pool( 4, 'providers' );

`Queuety::run_auto_scaling_worker_pool( $min_workers, $max_workers, $queue = 'default' )`

Напрямую запустить адаптивный пул рабочих процессов. Требования те же, что и для run_worker_pool().

PHP

Queuety::run_auto_scaling_worker_pool( 2, 6, 'providers' );

Объект-значение `Batch`

Метод/свойство	Тип	Описание
`$id`	`int`	Идентификатор группы
`$name`	`string\|null`	Необязательное имя группы
`$total_jobs`	`int`	Общее число задач в группе
`$pending_jobs`	`int`	Задачи, ожидающие выполнения
`$failed_jobs`	`int`	Число задач с ошибкой
`$failed_job_ids`	`array`	Идентификаторы задач с ошибкой
`$created_at`	`DateTimeImmutable`	Время создания группы
`$finished_at`	`DateTimeImmutable\|null`	Время завершения группы
`$cancelled_at`	`DateTimeImmutable\|null`	Время отмены группы
`progress()`	`int`	Процент выполнения (0–100)
`finished()`	`bool`	Завершена ли группа
`has_failures()`	`bool`	Есть ли задачи с ошибкой
`cancelled()`	`bool`	Отменена ли группа

`Queuety::chain( $jobs )`

Создаёт ChainBuilder для последовательного выполнения задач. Каждая задача зависит от успешного завершения предыдущей.

PHP

$first_job_id = Queuety::chain( [
    new FetchDataJob( $url ),
    new ProcessDataJob(),
    new NotifyCompleteJob(),
] )
    ->on_queue( 'pipeline' )
    ->catch( ChainFailedHandler::class )
    ->dispatch();

Методы `ChainBuilder`

Метод	Возвращает	Описание
`on_queue( string $queue )`	`self`	Задать очередь для всех задач в цепочке
`catch( string $handler_class )`	`self`	Обработчик, вызываемый, если любая задача в цепочке окончательно перемещена в хранилище неудачных задач
`dispatch()`	`int`	Создать все задачи с последовательными зависимостями; возвращает идентификатор первой задачи

`Queuety::retry( $job_id )`

Повторить выполнение конкретной задачи. Возвращает её в статус pending.

PHP

Queuety::retry( 42 );

`Queuety::retry_buried()`

Повторить выполнение всех перемещённых в хранилище неудачных задач. Возвращает число задач, отправленных на повтор.

PHP

$count = Queuety::retry_buried();

`Queuety::purge( $older_than_days )`

Удалить завершённые задачи. Возвращает число удалённых задач.

PHP

$count = Queuety::purge();     // uses QUEUETY_RETENTION_DAYS
$count = Queuety::purge( 30 ); // purge jobs older than 30 days

Очереди

`Queuety::stats( $queue )`

Получить число задач в группировке по статусу.

PHP

$stats = Queuety::stats();          // all queues
$stats = Queuety::stats( 'emails' ); // specific queue
// ['pending' => 12, 'processing' => 3, 'completed' => 150, 'failed' => 2, 'buried' => 1]

`Queuety::buried( $queue )`

Получить все задачи, перемещённые в хранилище неудачных задач.

PHP

$jobs = Queuety::buried();
$jobs = Queuety::buried( 'emails' );

`Queuety::pause( $queue )`

Приостановить очередь, чтобы рабочие процессы её пропускали.

PHP

Queuety::pause( 'emails' );

`Queuety::resume( $queue )`

Возобновить приостановленную очередь.

PHP

Queuety::resume( 'emails' );

`Queuety::is_paused( $queue )`

Проверяет, приостановлена ли очередь.

PHP

if ( Queuety::is_paused( 'emails' ) ) {
    // Queue is paused
}

Рабочие процессы

`Queuety::workflow( $name )`

Начинает создание рабочего процесса. Возвращает WorkflowBuilder.

PHP

$workflow_id = Queuety::workflow( 'generate_report' )
    ->then( FetchDataHandler::class )
    ->then( CallLLMHandler::class )
    ->then( FormatOutputHandler::class )
    ->dispatch( [ 'user_id' => 42 ] );

Методы `WorkflowBuilder`

Метод	Описание
`then( string $handler_class, ?string $name = null )`	Добавляет последовательный шаг. Необязательное имя используется для условного ветвления.
`parallel( array $handler_classes )`	Добавляет группу параллельных шагов.
`for_each( string $items_key, string $handler_class, ?string $result_key = null, ForEachMode $mode = ForEachMode::All, ?int $quorum = null, ?string $reducer_class = null, ?string $name = null )`	Добавляет динамический шаг for-each, создающий отдельную ветвь для каждого элемента в состоянии рабочего процесса.
`repeat_while( string $target_step, ?string $state_key = null, mixed $expected = true, ?string $name = null, ?string $condition_class = null, ?int $max_iterations = null )`	Возвращается к предыдущему именованному шагу, пока совпадает общедоступный ключ состояния или класс `RepeatCondition`.
`repeat_until( string $target_step, ?string $state_key = null, mixed $expected = true, ?string $name = null, ?string $condition_class = null, ?int $max_iterations = null )`	Возвращается к предыдущему именованному шагу, пока общедоступный ключ состояния или класс `RepeatCondition` не совпадет.
`run_workflow( string $workflow_name, WorkflowBuilder $sub_builder )`	Добавляет шаг запуска рабочего процесса.
`delay( int $seconds = 0, int $minutes = 0, int $hours = 0, int $days = 0 )`	Добавляет шаг долговременной задержки. Все длительности суммируются.
`wait_for_signal( string $name, ?string $result_key = null, ?string $step_name = null, array $match_payload = [], ?string $correlation_key = null )`	Добавляет шаг ожидания сигнала. Необязательные `match_payload` и `correlation_key` позволяют игнорировать посторонние события.
`wait_for_signals( array $signal_names, WaitMode $mode = WaitMode::All, ?string $result_key = null, ?string $name = null, array $match_payload = [], ?string $correlation_key = null )`	Ожидает все или любой из нескольких сигналов, с необязательным сопоставлением данных и корреляцией.
`wait_for_approval( string $signal_name = 'approval', ?string $result_key = 'approval', ?string $name = null, array $match_payload = [], ?string $correlation_key = null )`	Ожидает сигнал подтверждения от человека.
`wait_for_decision( string $approve_signal = 'approved', string $reject_signal = 'rejected', ?string $result_key = 'decision', ?string $name = null, array $match_payload = [], ?string $correlation_key = null )`	Ожидает подтверждения или отклонения и сохраняет структурированные данные решения.
`wait_for_input( string $signal_name = 'input', ?string $result_key = 'input', ?string $name = null, array $match_payload = [], ?string $correlation_key = null )`	Ожидает сигнал ввода от человека.
`wait_for_workflow( int\|string $workflow, ?string $result_key = null, ?string $name = null )`	Ожидает завершения одного рабочего процесса по ID или рабочего процесса с ID из общедоступного состояния.
`wait_for_workflows( array\|string $workflows, WaitMode $mode = WaitMode::All, ?string $result_key = null, ?string $name = null, ?int $quorum = null )`	Ожидает завершения всех, любого или кворума из нескольких рабочих процессов.
`wait_for_workflow_group( string $group_key, WaitMode $mode = WaitMode::All, ?int $quorum = null, ?string $result_key = null, ?string $name = null )`	Ожидает группу ранее запущенных рабочих процессов по внутреннему ключу группы.
`start_workflows( string $items_key, WorkflowBuilder $workflow_builder, string $result_key = 'started_workflow_ids', string $payload_key = 'item', bool $inherit_state = true, ?string $name = null, ?string $group_key = null )`	Запускает один рабочий процесс верхнего уровня для каждого элемента среды выполнения и при необходимости помечает запущенный набор долговечным ключом группы.
`start_agents( string $items_key, WorkflowBuilder $workflow_builder, string $result_key = 'agent_workflow_ids', string $payload_key = 'agent_task', bool $inherit_state = true, ?string $name = null, ?string $group_key = null )`	Псевдоним `start_workflows()` для сценариев с агентами.
`wait_for_agents( array\|string $workflows = 'agent_workflow_ids', WaitMode $mode = WaitMode::All, ?string $result_key = 'agent_results', ?string $name = null, ?int $quorum = null )`	Псевдоним `wait_for_workflows()` для сценариев с агентами.
`wait_for_agent_group( string $group_key = 'agents', WaitMode $mode = WaitMode::All, ?int $quorum = null, ?string $result_key = 'agent_results', ?string $name = null )`	Псевдоним `wait_for_workflow_group()` для сценариев с агентами.
`version( string $version )`	Помечает определение рабочего процесса версией уровня приложения для просмотра и экспорта.
`idempotency_key( string $key )`	Делает запуск рабочего процесса долговечным для указанного вызывающим кодом ключа. Повторный запуск с тем же ключом возвращает исходный ID рабочего процесса.
`max_transitions( int $max )`	Завершает рабочий процесс с ошибкой, если выполнено более `$max` переходов между шагами.
`max_for_each_items( int $max )`	Завершает рабочий процесс с ошибкой, если шаг `for_each()` пытается создать более `$max` элементов.
`max_state_bytes( int $bytes )`	Завершает рабочий процесс с ошибкой, если общедоступное состояние превышает `$bytes`.
`max_cost_units( int $max )`	Завершает рабочий процесс с ошибкой, если завершенные шаги расходуют более `$max` единиц стоимости.
`max_started_workflows( int $max )`	Завершает рабочий процесс с ошибкой, если он пытается запустить более `$max` дочерних рабочих процессов.
`compensate_with( string $handler_class )`	Назначает обработчик компенсации последнему добавленному шагу.
`compensate_on_failure()`	Автоматически запускает компенсацию завершенных шагов при сбое рабочего процесса.
`on_queue( string $queue )`	Задает очередь для всех шагов.
`with_priority( Priority $priority )`	Задает приоритет для всех шагов.
`max_attempts( int $max )`	Задает максимальное число повторных попыток для каждого шага.
`on_cancel( string $handler_class )`	Регистрирует класс обработчика очистки при отмене. См. Отмена.
`prune_state_after( int $steps = 2 )`	Включает автоматическую очистку состояния. Удаляет результаты шагов старше заданного порога. См. Очистка состояния.
`must_complete_within( int $seconds = 0, int $minutes = 0, int $hours = 0, int $days = 0 )`	Задает срок выполнения рабочего процесса. Все длительности суммируются. См. Сроки выполнения.
`on_deadline( string $handler_class )`	Регистрирует обработчик, вызываемый при превышении срока выполнения рабочего процесса. См. Сроки выполнения.
`dispatch( array $initial_payload = [] )`	Запускает рабочий процесс. Возвращает его ID.

`Queuety::workflow_status( $workflow_id )`

Получает статус рабочего процесса. Возвращает объект WorkflowState или null.

PHP

$state = Queuety::workflow_status( $workflow_id );
echo $state->name;          // 'generate_report'
echo $state->status->value; // 'running'
echo $state->current_step;  // 1
echo $state->total_steps;   // 3
print_r( $state->state );   // accumulated public state array
echo $state->wait_type;     // 'signal' or 'workflow' while blocked
echo $state->wait_mode;     // 'all', 'any', or 'quorum'
echo $state->definition_version; // 'research.v2'
echo $state->definition_hash;    // '4c7a...'
echo $state->idempotency_key;    // 'brief:42:research'
echo $state->artifact_count;     // 2
print_r( $state->budget );       // configured workflow budget + current usage
print_r( $state->wait_details ); // matched/remaining blockers, result key, correlation info
print_r( $state->artifact_keys ); // ['draft', 'citations']

WorkflowState также предоставляет:

waiting_for: имена сигналов или ID рабочих процессов, которые сейчас блокируют выполнение
current_step_name: настроенное имя текущего шага, если оно есть
wait_mode: текущий режим ожидания (all, any или quorum) во время блокировки
wait_details: дополнительные данные для просмотра, например, совпавшие и оставшиеся блокирующие условия
definition_version: значение, заданное с помощью version(), если оно есть
definition_hash: детерминированный хеш определения рабочего процесса
idempotency_key: долговечный ключ запуска, если он есть
budget: настроенные ограничения и текущее использование, если защитные ограничения включены
artifact_count: число сохраненных артефактов рабочего процесса
artifact_keys: ключи сохраненных артефактов для быстрой проверки

`Queuety::retry_workflow( $workflow_id )`

Повторно запускает рабочий процесс с ошибкой с шага, на котором произошел сбой.

PHP

Queuety::retry_workflow( $workflow_id );

Выбрасывает RuntimeException, если компенсация рабочего процесса после сбоя уже выполнена.

`Queuety::pause_workflow( $workflow_id )`

Приостанавливает выполняющийся рабочий процесс. Уже поставленные в очередь задачи текущего шага могут завершиться, но следующие шаги не ставятся в очередь до возобновления рабочего процесса.

PHP

Queuety::pause_workflow( $workflow_id );

`Queuety::resume_workflow( $workflow_id )`

Возобновляет приостановленный рабочий процесс.

PHP

Queuety::resume_workflow( $workflow_id );

`Queuety::cancel_workflow( $workflow_id )`

Отменяет рабочий процесс и запускает все зарегистрированные обработчики очистки. Устанавливает статус cancelled и перемещает все ожидающие и выполняющиеся задачи рабочего процесса в отложенные.

PHP

Queuety::cancel_workflow( $workflow_id );

Выбрасывает RuntimeException, если рабочий процесс уже completed или cancelled. Подробности см. в разделе Отмена.

`Queuety::rewind_workflow( $workflow_id, $to_step )`

Возвращает рабочий процесс к предыдущему шагу и повторно запускает его с этого места. Состояние восстанавливается из журнала событий state_after для указанного шага, после чего в очередь ставится следующий шаг.

PHP

Queuety::rewind_workflow( $workflow_id, to_step: 1 );

Выбрасывает RuntimeException, если для целевого шага нет сохраненного состояния. Требуется Журнал событий рабочего процесса. Подробности см. в разделе Перемотка времени.

`Queuety::fork_workflow( $workflow_id )`

Создает независимую копию выполняющегося рабочего процесса с его текущим состоянием. Возвращает ID нового рабочего процесса.

PHP

$forked_id = Queuety::fork_workflow( $workflow_id );

Созданный рабочий процесс получает собственный ID и выполняется независимо от исходного. Подробности см. в разделе Создание ветви.

`Queuety::export_workflow( $workflow_id )`

Экспортирует полную историю выполнения рабочего процесса в виде массива, сериализуемого в JSON. Включает строку рабочего процесса, все задачи, события рабочего процесса и записи журнала.

PHP

$data = Queuety::export_workflow( $workflow_id );

Описание структуры JSON и примеры использования см. в разделе Экспорт и повторное воспроизведение.

`Queuety::replay_workflow( $data )`

Повторно воспроизводит экспортированный рабочий процесс в текущей среде. Создает новый рабочий процесс на основе экспортированных данных и возвращает его ID.

PHP

$new_id = Queuety::replay_workflow( $data );

Подробности см. в разделе Экспорт и повторное воспроизведение.

`Queuety::replace_wp_cron()`

Заменяет встроенную систему WordPress cron планировщиком Queuety. Перехватывает wp_schedule_event(), wp_unschedule_event() и wp_get_scheduled_event() и направляет повторяющиеся события рабочему процессу Queuety.

PHP

Queuety::replace_wp_cron();

Работает только при загруженном WordPress. Подробности см. в разделе WP-Cron Замена.

`Queuety::restore_wp_cron()`

Восстанавливает поведение WordPress cron по умолчанию, удаляя связующие хуки Queuety cron.

PHP

Queuety::restore_wp_cron();

Ограничения описаны в разделе WP-Cron Замена.

`Queuety::signal( $workflow_id, $name, $data )`

Отправляет сигнал рабочему процессу. Если рабочий процесс ожидает этот сигнал, он немедленно возобновляется. В противном случае сигнал сохраняется и обрабатывается, когда рабочий процесс дойдет до соответствующего шага wait_for_signal.

PHP

Queuety::signal( $workflow_id, 'approved' );

// With data merged into workflow state
Queuety::signal( $workflow_id, 'approved', [
    'approved_by' => 'admin@example.com',
] );

Подробности см. в разделе Сигналы рабочего процесса.

`Queuety::approve_workflow( $workflow_id, $data = [], $signal_name = 'approval' )`

Семантическая обертка над signal() для сценариев подтверждения.

PHP

Queuety::approve_workflow( $workflow_id, [
    'reviewer' => 'editor@example.com',
] );

`Queuety::reject_workflow( $workflow_id, $data = [], $signal_name = 'rejected' )`

Семантическая обертка над signal() для сценариев отклонения.

PHP

Queuety::reject_workflow( $workflow_id, [
    'reason' => 'needs citations',
] );

`Queuety::submit_workflow_input( $workflow_id, $data = [], $signal_name = 'input' )`

Семантическая обертка над signal() для структурированного ввода человеком.

PHP

Queuety::submit_workflow_input( $workflow_id, [
    'notes' => 'Ship after legal review',
] );

`Queuety::put_artifact( $workflow_id, $artifact_key, $content, $kind = 'json', $step_index = null, $metadata = [] )`

Сохраняет один долговечный артефакт для рабочего процесса.

PHP

Queuety::put_artifact(
    $workflow_id,
    'research_brief',
    [ 'summary' => 'done' ],
    'json',
    metadata: [ 'source' => 'agent' ],
);

`Queuety::put_current_artifact( $artifact_key, $content, $kind = 'json', $metadata = [] )`

Сохраняет один артефакт для выполняемого в данный момент шага рабочего процесса.

PHP

Queuety::put_current_artifact(
    'draft',
    [ 'status' => 'ready' ],
    'json',
);

`Queuety::workflow_artifact( $workflow_id, $artifact_key )`

Извлекает один сохраненный артефакт рабочего процесса.

PHP

$artifact = Queuety::workflow_artifact( $workflow_id, 'draft' );

`Queuety::workflow_artifacts( $workflow_id, $include_content = false )`

Выводит список артефактов рабочего процесса.

PHP

$artifacts = Queuety::workflow_artifacts( $workflow_id );

`Queuety::delete_workflow_artifact( $workflow_id, $artifact_key )`

Удаляет один артефакт рабочего процесса.

PHP

Queuety::delete_workflow_artifact( $workflow_id, 'draft' );

Конечные автоматы

`Queuety::machine( $name )`

Начинает создание конечного автомата. Возвращает StateMachineBuilder.

PHP

use Queuety\Enums\StateMachineStatus;

$machine_id = Queuety::machine( 'agent_session' )
    ->state( 'awaiting_user' )
    ->on( 'user_message', 'planning' )
    ->state( 'planning' )
    ->action( PlanSessionAction::class )
    ->on( 'planned', 'completed' )
    ->state( 'completed', StateMachineStatus::Completed )
    ->dispatch( [ 'thread_id' => 42 ] );

Методы `StateMachineBuilder`

Метод	Описание
`initial( string $state_name )`	Явно задает начальное состояние.
`state( string $state_name, ?StateMachineStatus $terminal_status = null )`	Начинает настройку состояния.
`action( string\|array $action_class )`	Регистрирует действие входа в текущее состояние, помещаемое в очередь. Передайте имя класса или структурированное определение с `class`, `payload` и `config`; `payload` передается в `StateAction::handle()` как `$payload`.
`on( string $event, string $target, string\|array\|null $guard_class = null, ?string $name = null )`	Регистрирует переход по событию из текущего состояния. Условие принимает тот же структурированный формат, что и `action()`.
`on_queue( string $queue )`	Задает очередь для действий состояния, помещаемых в очередь.
`with_priority( Priority $priority )`	Задает приоритет действий состояния, помещаемых в очередь.
`max_attempts( int $max_attempts )`	Задает лимит повторных попыток для действий состояния, помещаемых в очередь.
`version( string $version )`	Помечает определение автомата версией уровня приложения.
`idempotency_key( string $key )`	Делает запуск долговечным для указанного вызывающим кодом ключа. Повторный запуск с тем же ключом возвращает исходный ID автомата.
`dispatch( array $initial_state = [] )`	Запускает автомат и возвращает его ID.

Классы действий входа в состояние также могут предоставлять необязательный метод config() с теми же ключами политики ресурсов, что поддерживаются шагами рабочего процесса: concurrency_group, concurrency_limit и cost_units.

`Queuety::dispatch_state_machine_definition( $definition, $initial_state = [], $options = [] )`

Запускает сериализованный пакет определения конечного автомата. В большинстве случаев в коде приложения следует предпочесть machine()->dispatch(). Этот низкоуровневый метод нужен адаптерам, чтобы запускать автомат без прямой зависимости от построителя.

PHP

$machine_id = Queuety::dispatch_state_machine_definition(
    $definition,
    [ 'thread_id' => 42 ],
    [ 'idempotency_key' => 'thread:42' ],
);

`Queuety::machine_status( $machine_id )`

Получает текущий статус автомата. Возвращает объект StateMachineState или null.

PHP

$status = Queuety::machine_status( $machine_id );

echo $status->name;               // 'agent_session'
echo $status->status->value;      // 'waiting_event'
echo $status->current_state;      // 'awaiting_review'
print_r( $status->state );        // public machine state
print_r( $status->available_events ); // valid incoming events in the current state

StateMachineState также предоставляет:

definition_version
definition_hash
idempotency_key
error_message
current_action
terminal_status

`Queuety::machine_event( $machine_id, $event_name, $payload = [] )`

Отправляет внешнее событие ожидающему автомату.

PHP

Queuety::machine_event(
    $machine_id,
    'approve',
    [
        'approved' => true,
        'reviewer' => 'editor@example.com',
    ]
);

Данные доступны условиям перехода и объединяются с общедоступным состоянием автомата после принятия события.

`Queuety::list_machines( $limit = 50, $status = null )`

Выводит список сохраненных записей автомата для проверки.

PHP

$machines = Queuety::list_machines();
$waiting  = Queuety::list_machines( status: 'waiting_event' );

Каждая запись содержит ID и имя автомата, статус, текущее состояние, версию определения, хеш определения, ключ идемпотентности и временные метки жизненного цикла.

`Queuety::machine_timeline( $machine_id, $limit = 100, $offset = 0 )`

Получает долговременную временную шкалу событий одного автомата.

PHP

$timeline = Queuety::machine_timeline( $machine_id );
$page_two = Queuety::machine_timeline( $machine_id, 100, 100 );

Каждая запись события содержит: id, machine_id, job_id, state_name, handler, event, event_name, transition_name, queue, attempt, input, output, state_before, state_after, context, artifacts, chunks, error, duration_ms и created_at.

`Queuety::machine_trace( $machine_id )`

Получает нормализованный пакет трассировки для отладчика конечного автомата UI.

PHP

$trace = Queuety::machine_trace( $machine_id );

Трассировка содержит machine, сгруппированные states, исходные events, связанные действия jobs, logs, artifacts и chunks.

`Queuety::machine_state_at( $machine_id, $event_id )`

Получает общедоступное состояние после определенного события трассировки автомата или null, если оно не найдено.

PHP

$state = Queuety::machine_state_at( $machine_id, $event_id );

Описание модели жизненного цикла и примеры использования см. в разделе Конечные автоматы.

Шаблоны рабочих процессов

`Queuety::define_workflow( $name )`

Создает построитель рабочего процесса для использования в качестве шаблона.

PHP

$builder = Queuety::define_workflow( 'onboarding' )
    ->then( CreateAccountHandler::class )
    ->then( SendWelcomeEmailHandler::class );

`Queuety::register_workflow_template( $builder )`

Регистрирует шаблон рабочего процесса.

PHP

Queuety::register_workflow_template( $builder );

`Queuety::run_workflow( $name, $payload, $options = [] )`

Запускает зарегистрированный шаблон рабочего процесса по имени. Возвращает ID рабочего процесса.

PHP

$workflow_id = Queuety::run_workflow( 'onboarding', [ 'email' => 'user@example.com' ] );

$workflow_id = Queuety::run_workflow(
    'onboarding',
    [ 'email' => 'user@example.com' ],
    [ 'idempotency_key' => 'onboarding:user@example.com' ],
);

Выбрасывает RuntimeException, если шаблон не зарегистрирован.

`Queuety::dispatch_workflow_definition( $definition, $payload, $options = [] )`

Запускает рабочий процесс из пакета определения среды выполнения.

В большинстве случаев в коде приложения следует предпочесть run_workflow() или workflow()->dispatch(). Этот низкоуровневый метод позволяет повторно используемым адаптерам запускать сериализованное определение рабочего процесса без зависимости от внутренних методов диспетчера рабочих процессов.

PHP

$workflow_id = Queuety::dispatch_workflow_definition(
    $definition,
    [ 'email' => 'user@example.com' ],
    [ 'idempotency_key' => 'onboarding:user@example.com' ],
);

`Queuety::on_action( $hook, $workflow, $map = null, $when = null, $idempotency_key = null, $priority = 10, $accepted_args = null )`

Регистрирует действие WordPress, запускающее рабочий процесс.

PHP

Queuety::on_action(
    'save_post',
    workflow: 'content_review',
    map: static fn ( int $post_id, object $post, bool $update ): array => [
        'post_id' => $post_id,
        'post_type' => $post->post_type,
        'update' => $update,
    ],
    when: static fn ( int $post_id, object $post ): bool => 'post' === $post->post_type,
    idempotency_key: static fn ( int $post_id ): string => "save_post:{$post_id}",
);

$workflow может быть:

именем зарегистрированного шаблона рабочего процесса
встроенным WorkflowBuilder

map, when и idempotency_key выполняются синхронно в рамках текущего запроса WordPress. После запуска асинхронно выполняются только шаги рабочего процесса.

Полное руководство см. в разделе Триггеры действий.

`Queuety::workflow_templates()`

Получает реестр шаблонов рабочих процессов.

PHP

$registry = Queuety::workflow_templates();
$template = $registry->get( 'onboarding' ); // WorkflowTemplate or null

`Queuety::load_workflows( $directory, $recursive )`

Загружает и регистрирует все файлы рабочих процессов из каталога. Каждый файл .php должен возвращать WorkflowBuilder. Возвращает число зарегистрированных рабочих процессов.

PHP

$count = Queuety::load_workflows( __DIR__ . '/workflows/' );
$count = Queuety::load_workflows( __DIR__ . '/workflows/', recursive: true );

`Queuety::load_workflow_file( $file_path )`

Загружает и регистрирует один файл рабочего процесса. Возвращает WorkflowTemplate или null, если файл не возвращает построитель.

PHP

$template = Queuety::load_workflow_file( __DIR__ . '/workflows/onboard-user.php' );

Планирование

`Queuety::schedule( $handler, $payload )`

Создает повторяющееся расписание. Возвращает построитель PendingSchedule.

PHP

Queuety::schedule( 'cleanup_handler' )
    ->every( '1 hour' )
    ->on_queue( 'maintenance' );

Queuety::schedule( 'nightly_report' )
    ->cron( '0 3 * * *' )
    ->on_queue( 'reports' );

Методы `PendingSchedule`

Метод	Описание
`every( string $interval )`	Задает выражение интервала (e.g. `1 hour`, `30 minutes`)
`cron( string $expression )`	Задает выражение cron из 5 полей
`overlap( OverlapPolicy $policy )`	Задает политику пересечения запусков (по умолчанию: `Allow`). См. Планирование.
`on_queue( string $queue )`	Задает целевую очередь (по умолчанию: `default`)
`id()`	Принудительно запускает и возвращает ID расписания

`Queuety::scheduler()`

Получает внутренний экземпляр Scheduler.

PHP

$scheduler = Queuety::scheduler();
$schedules = $scheduler->list();
$scheduler->remove( 'handler_name' );
$count = $scheduler->tick(); // manually trigger

Регистрация обработчиков

`Queuety::register( $name, $class )`

Регистрирует класс обработчика под указанным именем.

PHP

Queuety::register( 'send_email', SendEmailHandler::class );

`Queuety::discover_handlers( $directory, $namespace )`

Автоматически обнаруживает и регистрирует классы обработчиков из каталога. Возвращает число зарегистрированных обработчиков.

PHP

$count = Queuety::discover_handlers( __DIR__ . '/handlers', 'MyPlugin\\Handlers' );

`Queuety::registry()`

Получает реестр обработчиков.

PHP

$registry = Queuety::registry();

Наблюдаемость

`Queuety::logger()`

Получает экземпляр Logger для запроса записей журнала.

PHP

$logger = Queuety::logger();
$logs   = $logger->for_job( 42 );
$logs   = $logger->for_workflow( 7 );
$logs   = $logger->for_handler( 'send_email', 50 );
$logs   = $logger->for_event( LogEvent::Failed, 50 );
$logs   = $logger->since( new \DateTimeImmutable( '-1 hour' ), 100 );
$count  = $logger->purge( 30 );

`Queuety::metrics()`

Получает экземпляр Metrics.

PHP

$stats = Queuety::metrics()->handler_stats( 60 ); // last 60 minutes

`Queuety::webhook_notifier()`

Получает экземпляр WebhookNotifier.

PHP

$notifier = Queuety::webhook_notifier();
$id       = $notifier->register( 'job.buried', 'https://hooks.slack.com/...' );
$webhooks = $notifier->list();
$notifier->remove( $id );

`Queuety::rate_limiter()`

Получает экземпляр RateLimiter.

PHP

$limiter = Queuety::rate_limiter();
$limiter->register( 'call_openai', 60, 60 );

Кэш

`Queuety::cache()`

Получает экземпляр активного кэш-бэкенда.

PHP

$cache = Queuety::cache();
$cache->set( 'my_key', 'my_value', 60 );
$value = $cache->get( 'my_key' );

`Queuety::set_cache( $cache )`

Заменяет кэш-бэкенд. Вызовите перед Queuety::init(), чтобы использовать пользовательскую реализацию.

PHP

Queuety::set_cache( new RedisCache() );
Queuety::init( $connection );

`QUEUETY_CACHE_TTL`

TTL кэша по умолчанию в секундах (по умолчанию: 5). Чтобы изменить TTL, используемый внутренними операциями кэша по умолчанию, задайте эту константу в wp-config.php.

PHP

define( 'QUEUETY_CACHE_TTL', 10 );

Подробности о бэкендах и автоматическом определении см. в разделе Кэширование.

Журнал событий конечного автомата

`Queuety::machine_events()`

Получает экземпляр StateMachineEventLog.

PHP

$log = Queuety::machine_events();

Журнал событий рабочего процесса

`Queuety::workflow_events()`

Получает экземпляр WorkflowEventLog.

PHP

$log = Queuety::workflow_events();

`Queuety::workflow_timeline( $workflow_id, $limit = 100, $offset = 0 )`

Получает полную временную шкалу событий рабочего процесса. Возвращает массив записей событий, упорядоченных по ID.

PHP

$events = Queuety::workflow_timeline( $workflow_id );
$next   = Queuety::workflow_timeline( $workflow_id, 100, 100 );

Каждая строка события содержит: id, workflow_id, job_id, parent_event_id, step_index, step_name, step_type, handler, event, queue, attempt, input, output, state_before, state_after, context, artifacts, chunks, error, error_message (скопировано из error.message), duration_ms и created_at.

`Queuety::workflow_trace( $workflow_id )`

Получите нормализованный пакет трассировки для отладчика рабочих процессов UI.

PHP

$trace = Queuety::workflow_trace( $workflow_id );

Трассировка содержит workflow, сгруппированные steps, необработанные events, jobs, logs, artifacts, chunks, signals и wait_dependencies.

`Queuety::trace_input( $input )`

Задайте входные данные на уровне приложения для выполняемого в данный момент шага рабочего процесса.

`Queuety::trace_output( $output )`

Задайте выходные данные на уровне приложения для выполняемого в данный момент шага рабочего процесса.

`Queuety::trace_context( array $context )`

Объедините контекст на уровне приложения с выполняемым в данный момент шагом рабочего процесса.

`Queuety::workflow_state_at( $workflow_id, $step_index )`

Получите публичное состояние после завершения или возобновления определенного шага рабочего процесса либо null, если оно не найдено.

PHP

$state = Queuety::workflow_state_at( $workflow_id, 2 );

См. Журнал событий рабочего процесса, чтобы узнать о сценариях использования и командах CLI.

Внутренние экземпляры

Эти методы возвращают внутренние экземпляры для расширенных сценариев использования.

Метод	Возвращает	Описание
`Queuety::queue()`	`Queue`	Операции с очередью (получение, освобождение, помещение в архив)
`Queuety::worker()`	`Worker`	Управление процессами обработчиков
`Queuety::workflow_manager()`	`Workflow`	Оркестрация рабочих процессов
`Queuety::machines()`	`StateMachine`	Управление жизненным циклом конечного автомата
`Queuety::connection()`	`Connection`	Подключение к базе данных
`Queuety::batch_manager()`	`BatchManager`	Управление жизненным циклом пакетов (отмена, очистка)
`Queuety::chunk_store()`	`ChunkStore`	Сохранение фрагментов для потоковых шагов
`Queuety::machine_events()`	`StateMachineEventLog`	События трассировки конечного автомата и нормализованные пакеты трассировки
`Queuety::workflow_events()`	`WorkflowEventLog`	События трассировки рабочего процесса и нормализованные пакеты трассировки
`Queuety::resource_manager()`	`ResourceManager`	Профили допуска обработчиков, взвешенные лимиты и проверка системной памяти
`Queuety::cache()`	`Cache`	Экземпляр серверной части кэша

Тестирование

`Queuety::fake()`

Замените очередь имитацией в памяти для тестирования. Возвращает экземпляр QueueFake, который записывает отправленные задания, не обращаясь к базе данных. create_batch() также записывает метаданные имитации пакетов; цепочки проверяются по отдельным заданиям, которые они отправляют.

PHP

$fake = Queuety::fake();
SendEmailJob::dispatch( 'user@example.com', 'Hello', 'Hi!' );
$fake->assert_pushed( SendEmailJob::class );

`Queuety::reset()`

Сбросьте все внутреннее состояние, включая имитацию очереди. Вызывайте в tearDown().

PHP

Queuety::reset();

Методы `QueueFake`

Метод	Описание
`assert_pushed( string $class, ?Closure $callback )`	Проверить, что класс задания был отправлен. Необязательная функция обратного вызова фильтрует по данным задания.
`assert_pushed_times( string $class, int $count )`	Проверить, что задание было отправлено ровно N раз.
`assert_not_pushed( string $class )`	Проверить, что задание ни разу не отправлялось.
`assert_nothing_pushed()`	Проверить, что вообще не было отправлено ни одного задания.
`assert_batched( ?Closure $callback )`	Проверить, что пакет был отправлен. Необязательная функция обратного вызова фильтрует по данным пакета.
`pushed( string $class )`	Получить все записанные отправки для класса. Возвращает массив.
`batches()`	Получить все записанные пакеты. Возвращает массив.
`reset()`	Очистить все записанные задания и пакеты.

Подробные примеры см. в разделе Тестирование.

Интерфейсы

`Handler`

Интерфейс для простых обработчиков заданий.

PHP

interface Handler {
    public function handle( array $payload ): void;
    public function config(): array;
}

config() может возвращать queue, max_attempts, backoff, rate_limit, concurrency_group, concurrency_limit и cost_units.

`Step`

Интерфейс для обработчиков шагов рабочего процесса.

PHP

interface Step {
    public function handle( array $state ): array;
    public function config(): array;
}

config() может возвращать max_attempts, backoff, concurrency_group, concurrency_limit и cost_units.

`Contracts\ForEachHandler`

Интерфейс для обработчиков динамических ветвей цикла for-each.

PHP

namespace Queuety\Contracts;

interface ForEachHandler {
    public function handle_item( array $state, mixed $item, int $index ): array;
    public function config(): array;
}

config() может возвращать max_attempts, backoff, rate_limit, concurrency_group, concurrency_limit и cost_units.

`Contracts\ForEachReducer`

Необязательная функция сведения результатов for-each.

PHP

namespace Queuety\Contracts;

interface ForEachReducer {
    public function reduce( array $state, array $for_each ): array;
}

`Contracts\Compensation`

Интерфейс для обработчиков компенсации шагов.

PHP

namespace Queuety\Contracts;

interface Compensation {
    public function handle( array $state ): void;
}

`Contracts\RepeatCondition`

Интерфейс для предикатов циклов, используемых repeat_until() и repeat_while(), когда сравнения ключей состояния недостаточно.

PHP

namespace Queuety\Contracts;

interface RepeatCondition {
    public function matches( array $state ): bool;
}

`Contracts\StateAction`

Интерфейс для действий при входе в состояние, помещаемых в очередь.

PHP

namespace Queuety\Contracts;

interface StateAction {
    public function handle( array $state, ?string $event = null, array $event_payload = [], array $payload = [] ): array|string;
}

$payload передает структурированное значение payload из определения действия, если автомат был определен с ним. Возвращает обновления публичного состояния. Укажите _event и при необходимости _event_payload, если действие должно сразу вызвать переход.

Действия состояния также могут определять необязательный метод config(). Queuety считывает из него concurrency_group, concurrency_limit и cost_units при помещении действия входа в очередь.

`Contracts\StateGuard`

Интерфейс для условий перехода конечного автомата.

PHP

namespace Queuety\Contracts;

interface StateGuard {
    public function allows( array $state, array $event_payload, string $event, array $payload = [] ): bool;
}

`Contracts\Job`

Интерфейс для самостоятельных классов заданий, которые можно отправлять в очередь. Классы заданий инкапсулируют свои данные в публичных свойствах, а логику выполнения — в handle().

PHP

namespace Queuety\Contracts;

interface Job {
    public function handle(): void;
}

Классы заданий могут при необходимости определять метод middleware(), возвращающий массив экземпляров Contracts\Middleware.

`Contracts\Cache`

Интерфейс для серверных частей кэша. Встроенные реализации и пользовательские серверные части описаны в разделе Кэширование.

PHP

namespace Queuety\Contracts;

interface Cache {
    public function get( string $key ): mixed;
    public function set( string $key, mixed $value, int $ttl = 0 ): void;
    public function delete( string $key ): void;
    public function has( string $key ): bool;
    public function add( string $key, mixed $value, int $ttl = 0 ): bool;
    public function flush(): void;
}

Метод	Возвращает	Описание
`get( $key )`	`mixed`	Получить значение. Возвращает `null`, если значение не найдено или срок его действия истек.
`set( $key, $value, $ttl )`	`void`	Сохранить значение. TTL 0 означает, что срок действия не истекает.
`delete( $key )`	`void`	Удалить значение.
`has( $key )`	`bool`	Проверить, существует ли ключ и не истек ли срок его действия.
`add( $key, $value, $ttl )`	`bool`	Атомарно задать значение, только если ключ не существует. Возвращает `true`, если значение было задано.
`flush()`	`void`	Удалить все элементы из кэша.

`WorkflowEventLog`

Записывает переходы между шагами рабочего процесса вместе с полным контекстом трассировки. Доступ через Queuety::workflow_events().

Метод	Возвращает	Описание
`record_event( array $event )`	`void`	Записать необработанное событие трассировки
`record_step_started( ... )`	`void`	Записать событие step_started со входными данными и состоянием до выполнения
`record_step_completed( ... )`	`void`	Записать событие step_completed со входными и выходными данными, состоянием до и после выполнения
`record_step_failed( ... )`	`void`	Записать событие step_failed с контекстом трассировки и структурированными сведениями об ошибке
`get_timeline( int $workflow_id, ?int $limit = 100, int $offset = 0 )`	`array`	Получить события трассировки рабочего процесса по порядку, с возможностью постраничной навигации
`get_trace( int $workflow_id )`	`array`	Получить нормализованный пакет трассировки
`get_state_at_step( int $workflow_id, int $step_index )`	`array\|null`	Получить публичное состояние после завершения или возобновления определенного шага
`prune( int $older_than_days )`	`int`	Удалить события старше указанного числа дней. Возвращает количество удаленных строк.

`StateMachineEventLog`

Записывает события жизненного цикла конечного автомата, действий, переходов и условий с теми же полями трассировки, что и события рабочего процесса.

Метод	Возвращает	Описание
`record_event( array $event )`	`void`	Записать необработанное событие трассировки автомата
`get_timeline( int $machine_id, ?int $limit = 100, int $offset = 0 )`	`array`	Получить события трассировки автомата по порядку, с возможностью постраничной навигации
`get_trace( int $machine_id )`	`array`	Получить нормализованный пакет трассировки автомата
`get_state_at_event( int $machine_id, int $event_id )`	`array\|null`	Получить публичное состояние после определенного события трассировки автомата
`prune( int $older_than_days )`	`int`	Удалить события старше указанного числа дней. Возвращает количество удаленных строк.

`Contracts\StreamingStep`

Интерфейс для обработчиков шагов рабочего процесса, выдающих поток фрагментов. Каждое выданное значение немедленно сохраняется в базе данных. См. раздел Потоковые шаги.

PHP

namespace Queuety\Contracts;

interface StreamingStep {
    public function stream( array $state, array $existing_chunks = [] ): \Generator;
    public function on_complete( array $chunks, array $state ): array;
    public function config(): array;
}

Метод	Описание
`stream( $state, $existing_chunks )`	Генератор, выдающий строковые фрагменты. При каждой выдаче данные немедленно записываются в БД. При повторной попытке `$existing_chunks` содержит ранее сохраненные фрагменты.
`on_complete( $chunks, $state )`	Вызывается после завершения потока. Возвращает данные для объединения с состоянием рабочего процесса.
`config()`	Необязательная конфигурация. Поддерживает `max_attempts`, `backoff`, `concurrency_group`, `concurrency_limit` и `cost_units`.

`ChunkStore`

Управляет сохранением фрагментов для потоковых шагов. Доступ через Queuety::chunk_store().

Метод	Возвращает	Описание
`get_chunks( int $job_id )`	`string[]`	Получить все фрагменты задания, упорядоченные по индексу фрагмента
`append_chunk( int $job_id, int $chunk_index, string $content, ?int $workflow_id, ?int $step_index )`	`void`	Добавить один фрагмент
`clear_chunks( int $job_id )`	`void`	Удалить все фрагменты задания
`chunk_count( int $job_id )`	`int`	Подсчитать фрагменты задания
`get_accumulated( int $job_id )`	`string`	Объединить все фрагменты в одну строку

`Heartbeat`

Статический вспомогательный метод для отправки сигналов активности из обработчиков шагов и потоковых шагов. См. раздел Сигналы активности.

Метод	Описание
`Heartbeat::init( int $job_id, Connection $conn )`	Задать контекст задания. Вызывается обработчиком автоматически.
`Heartbeat::beat( array $progress = [] )`	Отправить сигнал активности. Обновляет `reserved_at` и при необходимости сохраняет данные о ходе выполнения в `heartbeat_data`.
`Heartbeat::clear()`	Очистить контекст задания. Вызывается обработчиком автоматически.
`Heartbeat::current_job_id()`	Получить текущий ID задания (для тестирования).

`Contracts\Middleware`

Интерфейс для промежуточного ПО заданий. Промежуточное ПО оборачивает выполнение задания в конвейер по принципу луковицы.

PHP

namespace Queuety\Contracts;

interface Middleware {
    public function handle( object $job, \Closure $next ): void;
}

Трейты

`Dispatchable`

Предоставляет статические методы отправки для классов заданий, реализующих Contracts\Job. Аргументы конструктора передаются при создании экземпляра задания.

PHP

use Queuety\Contracts\Job;
use Queuety\Dispatchable;

class SendEmailJob implements Job {
    use Dispatchable;
    // ...
}

// Dispatches a new SendEmailJob instance
SendEmailJob::dispatch( 'user@example.com', 'Subject', 'Body' );

Метод	Возвращает	Описание
`dispatch( ...$args )`	`PendingJob`	Создать экземпляр и отправить его
`dispatch_if( bool $condition, ...$args )`	`PendingJob\|null`	Отправить, только если `$condition` равно true
`dispatch_unless( bool $condition, ...$args )`	`PendingJob\|null`	Отправить, только если `$condition` равно false
`with_chain( array $jobs )`	`ChainBuilder`	Запустить цепочку от этого класса задания

PHP

// Conditional dispatch
SendEmailJob::dispatch_if( $user->wants_email, $user->email, 'Hello', 'Hi!' );
SendEmailJob::dispatch_unless( $maintenance_mode, $user->email, 'Hello', 'Hi!' );

// Chain from a job class
FetchDataJob::with_chain( [
    new ProcessDataJob(),
    new NotifyCompleteJob(),
] )->dispatch();

Свойства задания

Классы заданий могут объявлять публичные свойства, которые обработчик считывает с помощью рефлексии, чтобы переопределить параметры повторных попыток и времени ожидания.

Свойство	Тип	Описание
`$tries`	`int`	Максимальное число попыток, после которого задание помещается в архив
`$timeout`	`int`	Максимальное время выполнения в секундах (требуется `pcntl`)
`$backoff`	`array`	Увеличивающиеся задержки в секундах для каждой повторной попытки

PHP

class ImportProductsJob implements Job {
    use Dispatchable;

    public int $tries = 5;
    public int $timeout = 120;
    public array $backoff = [ 10, 30, 60, 120 ];

    // ...
}

Метод-перехватчик `failed()`

Когда задание помещается в архив после исчерпания всех повторных попыток, обработчик вызывает метод failed() экземпляра задания, если он определен.

PHP

public function failed( \Throwable $exception ): void {
    error_log( "Job failed: {$exception->getMessage()}" );
}

Встроенное промежуточное ПО

Все классы промежуточного ПО реализуют Contracts\Middleware и находятся в пространстве имен Queuety\Middleware.

Класс	Описание
`RateLimited( int $max, int $window )`	Ограничить частоту до `$max` выполнений за `$window` секунд
`Timeout( int $seconds )`	Завершить задание через `$seconds` секунд (требуется `pcntl`)
`UniqueJob( ?string $key )`	Предотвратить параллельное выполнение с помощью блокировки базы данных
`WithoutOverlapping( string $key, int $release_after )`	Предотвратить одновременное выполнение с автоматическим снятием блокировки по истечении срока
`ThrottlesExceptions( int $max_attempts, int $decay_minutes )`	Ограничить частоту исключений, чтобы предотвратить шквал заданий при недоступности внешней службы

Сведения об использовании и примеры см. в разделе Промежуточное ПО.

Перечисления

`Priority`

PHP

enum Priority: int {
    case Low    = 0;
    case Normal = 1;
    case High   = 2;
    case Urgent = 3;
}

`JobStatus`

PHP

enum JobStatus: string {
    case Pending    = 'pending';
    case Processing = 'processing';
    case Completed  = 'completed';
    case Failed     = 'failed';
    case Buried     = 'buried';
}

`WorkflowStatus`

PHP

enum WorkflowStatus: string {
    case Running        = 'running';
    case Completed      = 'completed';
    case Failed         = 'failed';
    case Paused         = 'paused';
    case WaitingForSignal  = 'waiting_for_signal';
    case WaitingForWorkflows = 'waiting_for_workflows';
    case Cancelled      = 'cancelled';
}

`StateMachineStatus`

PHP

enum StateMachineStatus: string {
    case Running = 'running';
    case WaitingEvent = 'waiting_event';
    case Completed = 'completed';
    case Failed = 'failed';
    case Paused = 'paused';
    case Cancelled = 'cancelled';
}

`BackoffStrategy`

PHP

enum BackoffStrategy: string {
    case Fixed       = 'fixed';
    case Linear      = 'linear';
    case Exponential = 'exponential';
}

`ForEachMode`

PHP

enum ForEachMode: string {
    case All = 'all';
    case FirstSuccess = 'first_success';
    case Quorum = 'quorum';
}

`WaitMode`

PHP

enum WaitMode: string {
    case All    = 'all';
    case Any    = 'any';
    case Quorum = 'quorum';
}

`LogEvent`

PHP

enum LogEvent: string {
    case Started                  = 'started';
    case Completed                = 'completed';
    case Failed                   = 'failed';
    case Buried                   = 'buried';
    case Retried                  = 'retried';
    case WorkflowStarted          = 'workflow_started';
    case WorkflowCompleted        = 'workflow_completed';
    case WorkflowFailed           = 'workflow_failed';
    case WorkflowPaused           = 'workflow_paused';
    case WorkflowResumed          = 'workflow_resumed';
    case WorkflowCancelled        = 'workflow_cancelled';
    case WorkflowRewound          = 'workflow_rewound';
    case WorkflowForked           = 'workflow_forked';
    case WorkflowDeadlineExceeded = 'workflow_deadline_exceeded';
    case Debug                    = 'debug';
}

`OverlapPolicy`

PHP

enum OverlapPolicy: string {
    case Allow  = 'allow';
    case Skip   = 'skip';
    case Buffer = 'buffer';
}

Подробнее о работе каждой политики см. в разделе Планирование.

Атрибуты

`#[QueuetyHandler]`

Атрибут PHP 8 для автоматической регистрации. См. раздел PHP Атрибуты.

PHP

use Queuety\Attributes\QueuetyHandler;

#[QueuetyHandler( name: 'send_email', queue: 'emails', max_attempts: 5 )]
class SendEmailHandler implements Handler {
    // ...
}

Мост CLI

`Queuety::cli_command()`

Возвращает опубликованное имя корневой команды WP-CLI.

Корневая команда по умолчанию — queuety. Задайте QUEUETY_CLI_COMMAND до запуска Queuety, если команду нужно опубликовать под другим именем.

`Queuety::cli_command_map()`

Возвращает сериализуемый каталог команд Queuety CLI для тестовых оболочек, которым нужно знать доступные команды без анализа классов WP-CLI.

Каждая запись содержит:

стабильный ID операции
относительный путь CLI
класс и метод обработчика WP-CLI
возможные цели выполнения
адаптер, нормализующий необработанные аргументы CLI в план выполнения

`Queuety::resolve_cli_command( $path, $args, $assoc_args )`

Преобразует разобранный вызов CLI в план выполнения, который должна запустить тестовая оболочка.

PHP

$plan = Queuety::resolve_cli_command(
    [ 'workflow', 'export' ],
    [ '42' ],
    [ 'output' => '/tmp/workflow.json' ]
);

echo $plan['callable']; // Queuety\Queuety::export_workflow_to_file
print_r( $plan['arguments'] );

Планы выполнения учитывают особенности транспорта:

команды разрешаются в публичные вызываемые объекты PHP с нормализованными аргументами
каталог заранее перечисляет все возможные цели выполнения, а callable и transport указывают ветвь, выбранную для разобранных входных данных

Так обычно используют Queuety в тестовых оболочках агентов: стабильный ID операции, публичная цель API и нормализация адаптером специфичных для CLI аспектов, таких как параметры очереди по умолчанию, декодирование данных JSON, выбор ветви пула обработчиков и обработка файлов экспорта рабочего процесса.

PHP API

Доступ к автономному ядру

Queuety::init( Connection $conn )

Queuety::ensure_schema()

Задачи

Queuety::dispatch( $handler, $payload )

Методы PendingJob

Queuety::dispatch_sync( $handler, $payload )

Queuety::batch( $jobs )

Queuety::create_batch( $jobs )

Методы BatchBuilder

Queuety::find_batch( $id )

Управление рабочими процессами

Queuety::run_worker( $queue = 'default', $once = false )

Queuety::run_worker_pool( $workers, $queue = 'default' )

Queuety::run_auto_scaling_worker_pool( $min_workers, $max_workers, $queue = 'default' )

Объект-значение Batch

Queuety::chain( $jobs )

Методы ChainBuilder

Queuety::retry( $job_id )

Queuety::retry_buried()

Queuety::purge( $older_than_days )

Очереди

Queuety::stats( $queue )

Queuety::buried( $queue )

Queuety::pause( $queue )

Queuety::resume( $queue )

Queuety::is_paused( $queue )

Рабочие процессы

Queuety::workflow( $name )

Методы WorkflowBuilder

Queuety::workflow_status( $workflow_id )

Queuety::retry_workflow( $workflow_id )

Queuety::pause_workflow( $workflow_id )

Queuety::resume_workflow( $workflow_id )

Queuety::cancel_workflow( $workflow_id )

Queuety::rewind_workflow( $workflow_id, $to_step )

Queuety::fork_workflow( $workflow_id )

Queuety::export_workflow( $workflow_id )

Queuety::replay_workflow( $data )

Queuety::replace_wp_cron()

Queuety::restore_wp_cron()

Queuety::signal( $workflow_id, $name, $data )

Queuety::approve_workflow( $workflow_id, $data = [], $signal_name = 'approval' )

Queuety::reject_workflow( $workflow_id, $data = [], $signal_name = 'rejected' )

Queuety::submit_workflow_input( $workflow_id, $data = [], $signal_name = 'input' )

Queuety::put_artifact( $workflow_id, $artifact_key, $content, $kind = 'json', $step_index = null, $metadata = [] )

Queuety::put_current_artifact( $artifact_key, $content, $kind = 'json', $metadata = [] )

Queuety::workflow_artifact( $workflow_id, $artifact_key )

Queuety::workflow_artifacts( $workflow_id, $include_content = false )

Queuety::delete_workflow_artifact( $workflow_id, $artifact_key )