PHP API

Преглед като Markdown Редактиране в GitHub Документация за 2cc7b2b

Всичко в Queuety е достъпно чрез статичния клас Queuety\Queuety. Това е единната входна точка за изпращане на задачи, изграждане на работни потоци, управление на машини на състоянията, управление на опашки и достъп до вътрешните подсистеми.

PHP

use Queuety\Queuety;

Достъп до самостоятелното ядро

Средата за изпълнение на Queuety, базирана на DB, може да се инициализира извън WordPress.

Този самостоятелен режим е предназначен за ядрото за оркестрация:

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

Това не замества адаптера WordPress. Queuety::on_action(), replace_wp_cron(), началното зареждане на плъгина и реалният път за регистрация на wp queuety ... все още изискват WordPress.

`Queuety::init( Connection $conn )`

Инициализирайте Queuety с директна връзка, базирана на DB.

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 от имената на runtime таблиците за всяка връзка:

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()`

Уверете се, че runtime таблиците 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 )`	Добавя задачата към споделена група за допускане на workers
`cost_units( int $units )`	Прикрепя метаданни за относителната цена на изпълнение към изпращането
`id()`	Принудително изпраща задачата и връща нейния ID

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

`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() на задачата се извиква директно. Не се създава ред в базата данни, не се включва worker и не се прилага междинен софтуер.

`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 )`

Намира група по ID. Връща обект-стойност Batch или null.

PHP

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

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

Управление на workers

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

Изпълнява директно един цикъл на worker.

PHP

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

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

Изпълнява директно пул от workers с фиксиран размер. Изисква разширенията 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' )`

Изпълнява директно адаптивен пул от workers. Изискванията са същите като за run_worker_pool().

PHP

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

Обект-стойност `Batch`

Метод/свойство	Тип	Описание
`$id`	`int`	ID на пакетната задача
`$name`	`string\|null`	Незадължително име на пакетната задача
`$total_jobs`	`int`	Общ брой задачи в пакетната задача
`$pending_jobs`	`int`	Задачи, които все още чакат изпълнение
`$failed_jobs`	`int`	Брой неуспешни задачи
`$failed_job_ids`	`array`	ID на неуспешните задачи
`$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`	Създава всички задачи с последователни зависимости и връща ID на първата задача

`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 )`	Добавя динамична стъпка за всяка стойност, която разгръща по един клон на задача за всеки елемент в състоянието на работния поток.
`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 )`	Задава 5-полеви израз cron
`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). Задайте тази константа в wp-config.php, за да промените TTL по подразбиране, използван от вътрешните кеш операции.

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 )`	Генератор, който генерира низови части. При всяко генериране се извършва незабавен запис в DB. При повторен опит `$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 )`	Ограничaва изключенията, за да предотврати лавинообразно стартиране на задачи при отказ на външна услуга

Вижте Мидълуер за подробности за употребата и примери.

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

`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 )

Управление на workers

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 )