PHP API
Vše v Queuety je dostupné prostřednictvím statické třídy Queuety\Queuety. Jde o jediné vstupní místo pro spouštění úloh, sestavování workflow, řízení stavových automatů, správu front a přístup k interním subsystémům.
use Queuety\Queuety;
Samostatný přístup k jádru
Běhové prostředí Queuety využívající databázi lze inicializovat mimo WordPress.
Tato samostatná cesta je určena pro jádro orchestrace:
- úlohy a workeři (skupiny workerů navíc potřebují konstanty
DB_HOST,DB_NAME,DB_USERaDB_PASSWORD, viz níže) - workflow a signály
- stavové automaty a události
- artefakty, protokoly, metriky a plány
Nenahrazuje adaptér WordPress. Queuety::on_action(),
replace_wp_cron(), inicializace pluginu a skutečná registrační cesta wp queuety ...
stále vyžadují WordPress.
Queuety::init( Connection $conn )
Inicializujte Queuety přímým připojením k databázi.
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 );
Volitelný argument table_prefix mění část Queuety názvů tabulek běhového prostředí pro každé připojení:
$conn = new Connection(
host: '127.0.0.1:3306',
dbname: 'wordpress',
user: 'root',
password: 'secret',
prefix: 'wp_',
table_prefix: 'themequeue_',
);
Díky tomu se tabulky jako queuety_jobs převedou na wp_themequeue_jobs.
Queuety::ensure_schema()
Zajistěte, aby pro aktivní připojení existovaly tabulky běhového prostředí Queuety.
Queuety::init( $conn );
Queuety::ensure_schema();
Úplný model a hranici WordPress najdete v části Samostatné použití.
Úlohy
Queuety::dispatch( $handler, $payload )
Spustí úlohu. Přijímá buď řetězec s názvem obslužné rutiny, nebo instanci Contracts\Job. Vrací objekt PendingJob pro řetězení možností.
// 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();
Je-li předána instance Contracts\Job, serializátor získá FQCN jako název obslužné rutiny a veřejné vlastnosti jako data. Parametr $payload se ignoruje, pokud je $handler instancí Job.
Metody PendingJob
| Metoda | Popis |
|---|---|
on_queue( string $queue ) |
Nastavit cílovou frontu |
with_priority( Priority $priority ) |
Nastavit úroveň priority |
delay( int $seconds ) |
Zpoždění, než bude úloha dostupná |
max_attempts( int $max ) |
Maximální počet pokusů o opakování |
rate_limit( int $max, int $window ) |
Omezení rychlosti: maximum spuštění za interval v sekundách |
unique() |
Zabránit duplicitnímu spuštění |
after( int $job_id ) |
Počkat, až se nejprve dokončí jiná úloha |
concurrency_group( string $group, int $limit ) |
Zařadit úlohu do sdílené skupiny pro přijímání workerů |
cost_units( int $units ) |
Připojit ke spuštění metadata o relativních nákladech na provedení |
id() |
Vynutit spuštění a vrátit ID úlohy |
concurrency_group() a cost_units() jsou součástí širšího modelu Správy prostředků, takže se stejné spuštění může účastnit limitů souběžnosti podle počtu, vážených rozpočtů front nebo poskytovatelů i heuristik pro přijímání workerů.
Queuety::dispatch_sync( $handler, $payload )
Provést úlohu synchronně v aktuálním procesu bez přidání do fronty.
// 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' ] );
Metoda úlohy handle() se zavolá přímo. Nevytvoří se žádný řádek v databázi, nepoužije se worker a middleware se neaplikuje.
Queuety::batch( $jobs )
Spustit více úloh v jediném vícerádkovém příkazu INSERT.
$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 )
Vytvořte BatchBuilder pro spuštění skupiny úloh se zpětnými voláními a sledováním průběhu. Podrobnosti najdete v části Dávkování.
$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();
Metody BatchBuilder
| Metoda | Vrací | Popis |
|---|---|---|
name( string $name ) |
self |
Nastavit název dávky |
on_queue( string $queue ) |
self |
Nastavit frontu pro všechny úlohy v dávce |
then( string $handler_class ) |
self |
Obslužná rutina volaná po úspěšném dokončení dávky |
catch( string $handler_class ) |
self |
Obslužná rutina volaná při selhání první úlohy v dávce |
finally( string $handler_class ) |
self |
Obslužná rutina volaná po dokončení dávky (úspěchu či selhání) |
allow_failures() |
self |
Povolit spuštění zpětného volání then i při selhání |
dispatch() |
Batch |
Vytvořit řádek dávky a všechny úlohy; vrátí objekt Batch |
Queuety::find_batch( $id )
Najít dávku podle ID. Vrací objekt Batch nebo null.
$batch = Queuety::find_batch( $batch_id );
if ( null !== $batch ) {
echo $batch->progress() . '% complete';
}
Řízení workerů
Queuety::run_worker( $queue = 'default', $once = false )
Spustit přímo jednu smyčku workera.
Queuety::run_worker( 'default' );
Queuety::run_worker( 'critical,default', true );
Queuety::run_worker_pool( $workers, $queue = 'default' )
Spustit přímo skupinu workerů pevné velikosti. Vyžaduje rozšíření pcntl a posix. Skupina pro svá podřízená připojení čte konstanty WordPress DB_HOST, DB_NAME, DB_USER a DB_PASSWORD (pouze předpony tabulek přebírá z inicializovaného připojení), proto je definujte, když skupinu spouštíte mimo WordPress (#5).
Queuety::run_worker_pool( 4, 'providers' );
Queuety::run_auto_scaling_worker_pool( $min_workers, $max_workers, $queue = 'default' )
Spustit přímo adaptivní skupinu workerů. Platí stejné požadavky jako pro run_worker_pool().
Queuety::run_auto_scaling_worker_pool( 2, 6, 'providers' );
Objekt hodnoty Batch
| Metoda/vlastnost | Typ | Popis |
|---|---|---|
$id |
int |
ID dávky |
$name |
string|null |
Volitelný název dávky |
$total_jobs |
int |
Celkový počet úloh v dávce |
$pending_jobs |
int |
Zbývající úlohy |
$failed_jobs |
int |
Počet neúspěšných úloh |
$failed_job_ids |
array |
ID neúspěšných úloh |
$created_at |
DateTimeImmutable |
Kdy byla dávka vytvořena |
$finished_at |
DateTimeImmutable|null |
Kdy byla dávka dokončena |
$cancelled_at |
DateTimeImmutable|null |
Kdy byla dávka zrušena |
progress() |
int |
Procento dokončení (0-100) |
finished() |
bool |
Zda byla dávka dokončena |
has_failures() |
bool |
Zda některá úloha selhala |
cancelled() |
bool |
Zda byla dávka zrušena |
Queuety::chain( $jobs )
Vytvoří ChainBuilder pro postupné spouštění úloh. Každá úloha závisí na úspěšném dokončení předchozí úlohy.
$first_job_id = Queuety::chain( [
new FetchDataJob( $url ),
new ProcessDataJob(),
new NotifyCompleteJob(),
] )
->on_queue( 'pipeline' )
->catch( ChainFailedHandler::class )
->dispatch();
Metody ChainBuilder
| Metoda | Vrací | Popis |
|---|---|---|
on_queue( string $queue ) |
self |
Nastaví frontu pro všechny úlohy v řetězci |
catch( string $handler_class ) |
self |
Obslužná rutina volaná, když je některá úloha v řetězci trvale přesunuta do úložiště neúspěšných úloh |
dispatch() |
int |
Vytvoří všechny úlohy se sekvenčními závislostmi a vrátí ID první úlohy |
Queuety::retry( $job_id )
Opakuje konkrétní úlohu. Obnoví její stav na pending.
Queuety::retry( 42 );
Queuety::retry_buried()
Opakuje všechny odložené úlohy. Vrací počet opakovaných úloh.
$count = Queuety::retry_buried();
Queuety::purge( $older_than_days )
Vymaže dokončené úlohy. Vrací počet vymazaných úloh.
$count = Queuety::purge(); // uses QUEUETY_RETENTION_DAYS
$count = Queuety::purge( 30 ); // purge jobs older than 30 days
Fronty
Queuety::stats( $queue )
Získá počty úloh seskupené podle stavu.
$stats = Queuety::stats(); // all queues
$stats = Queuety::stats( 'emails' ); // specific queue
// ['pending' => 12, 'processing' => 3, 'completed' => 150, 'failed' => 2, 'buried' => 1]
Queuety::buried( $queue )
Získá všechny odložené úlohy.
$jobs = Queuety::buried();
$jobs = Queuety::buried( 'emails' );
Queuety::pause( $queue )
Pozastaví frontu, aby ji pracovní procesy přeskočily.
Queuety::pause( 'emails' );
Queuety::resume( $queue )
Obnoví pozastavenou frontu.
Queuety::resume( 'emails' );
Queuety::is_paused( $queue )
Zjistí, zda je fronta pozastavená.
if ( Queuety::is_paused( 'emails' ) ) {
// Queue is paused
}
Pracovní postupy
Queuety::workflow( $name )
Začne vytvářet pracovní postup. Vrací WorkflowBuilder.
$workflow_id = Queuety::workflow( 'generate_report' )
->then( FetchDataHandler::class )
->then( CallLLMHandler::class )
->then( FormatOutputHandler::class )
->dispatch( [ 'user_id' => 42 ] );
Metody WorkflowBuilder
| Metoda | Popis |
|---|---|
then( string $handler_class, ?string $name = null ) |
Přidá sekvenční krok. Volitelný název slouží k podmíněnému větvení. |
parallel( array $handler_classes ) |
Přidá skupinu paralelních kroků. |
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 ) |
Přidá dynamický krok „pro každou položku“, který pro každou položku ve stavu pracovního postupu vytvoří jednu větevní úlohu. |
repeat_while( string $target_step, ?string $state_key = null, mixed $expected = true, ?string $name = null, ?string $condition_class = null, ?int $max_iterations = null ) |
Vrátí se k dřívějšímu pojmenovanému kroku, dokud odpovídá veřejný klíč stavu nebo třída RepeatCondition. |
repeat_until( string $target_step, ?string $state_key = null, mixed $expected = true, ?string $name = null, ?string $condition_class = null, ?int $max_iterations = null ) |
Vrátí se k dřívějšímu pojmenovanému kroku, dokud veřejný klíč stavu nebo třída RepeatCondition neodpovídá. |
run_workflow( string $workflow_name, WorkflowBuilder $sub_builder ) |
Přidá krok pro spuštění pracovního postupu. |
delay( int $seconds = 0, int $minutes = 0, int $hours = 0, int $days = 0 ) |
Přidá krok s trvalou prodlevou. Sčítají se všechny doby trvání. |
wait_for_signal( string $name, ?string $result_key = null, ?string $step_name = null, array $match_payload = [], ?string $correlation_key = null ) |
Přidá krok čekající na signál. Volitelné match_payload a correlation_key umožňují ignorovat nesouvisející události. |
wait_for_signals( array $signal_names, WaitMode $mode = WaitMode::All, ?string $result_key = null, ?string $name = null, array $match_payload = [], ?string $correlation_key = null ) |
Čeká na všechny nebo některé z několika signálů, případně s porovnáním dat a korelací. |
wait_for_approval( string $signal_name = 'approval', ?string $result_key = 'approval', ?string $name = null, array $match_payload = [], ?string $correlation_key = null ) |
Čeká na signál lidského schválení. |
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 ) |
Čeká na schválení nebo zamítnutí a uloží strukturovaná data rozhodnutí. |
wait_for_input( string $signal_name = 'input', ?string $result_key = 'input', ?string $name = null, array $match_payload = [], ?string $correlation_key = null ) |
Čeká na signál lidského vstupu. |
wait_for_workflow( int|string $workflow, ?string $result_key = null, ?string $name = null ) |
Čeká na dokončení jednoho pracovního postupu nebo pracovního postupu určeného ID uloženým ve veřejném stavu. |
wait_for_workflows( array|string $workflows, WaitMode $mode = WaitMode::All, ?string $result_key = null, ?string $name = null, ?int $quorum = null ) |
Čeká na dokončení všech, některých nebo kvóra z několika pracovních postupů. |
wait_for_workflow_group( string $group_key, WaitMode $mode = WaitMode::All, ?int $quorum = null, ?string $result_key = null, ?string $name = null ) |
Čeká na dříve spuštěnou skupinu pracovních postupů určenou interním klíčem skupiny. |
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 ) |
Spustí jeden pracovní postup nejvyšší úrovně pro každou položku za běhu a případně označí spuštěnou skupinu trvalým klíčem skupiny. |
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 ) |
Alternativní název start_workflows() určený pro agenty. |
wait_for_agents( array|string $workflows = 'agent_workflow_ids', WaitMode $mode = WaitMode::All, ?string $result_key = 'agent_results', ?string $name = null, ?int $quorum = null ) |
Alternativní název wait_for_workflows() určený pro agenty. |
wait_for_agent_group( string $group_key = 'agents', WaitMode $mode = WaitMode::All, ?int $quorum = null, ?string $result_key = 'agent_results', ?string $name = null ) |
Alternativní název wait_for_workflow_group() určený pro agenty. |
version( string $version ) |
Označí definici pracovního postupu verzí na úrovni aplikace pro kontrolu a export. |
idempotency_key( string $key ) |
Zajistí trvalé odesílání pracovního postupu pomocí klíče zadaného volajícím. Opětovné odeslání se stejným klíčem vrátí původní ID pracovního postupu. |
max_transitions( int $max ) |
Selže, pokud se dokončí více než $max přechodů mezi kroky. |
max_for_each_items( int $max ) |
Selže, pokud se krok for_each() pokusí rozšířit na více než $max položek. |
max_state_bytes( int $bytes ) |
Selže, pokud veřejný stav překročí velikost $bytes. |
max_cost_units( int $max ) |
Selže, pokud dokončené kroky spotřebují více než $max jednotek nákladů. |
max_started_workflows( int $max ) |
Selže dříve, než odešle více než $max podřízených pracovních postupů. |
compensate_with( string $handler_class ) |
Připojí obslužnou rutinu kompenzace k naposledy přidanému kroku. |
compensate_on_failure() |
Při selhání pracovního postupu automaticky spustí kompenzace dokončených kroků. |
on_queue( string $queue ) |
Nastaví frontu pro všechny kroky. |
with_priority( Priority $priority ) |
Nastaví prioritu pro všechny kroky. |
max_attempts( int $max ) |
Nastaví maximální počet opakovaných pokusů pro každý krok. |
on_cancel( string $handler_class ) |
Zaregistruje třídu obslužné rutiny pro úklid při zrušení. Viz Zrušení. |
prune_state_after( int $steps = 2 ) |
Povolí automatické prořezávání stavu. Odebere výstupy kroků starší než stanovený limit. Viz Prořezávání stavu. |
must_complete_within( int $seconds = 0, int $minutes = 0, int $hours = 0, int $days = 0 ) |
Nastaví lhůtu pracovního postupu. Sčítají se všechny doby trvání. Viz Lhůty. |
on_deadline( string $handler_class ) |
Zaregistruje obslužnou rutinu volanou při překročení lhůty pracovního postupu. Viz Lhůty. |
dispatch( array $initial_payload = [] ) |
Odešle pracovní postup. Vrací jeho ID. |
Queuety::workflow_status( $workflow_id )
Získá stav pracovního postupu. Vrací objekt WorkflowState nebo null.
$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 zpřístupňuje také:
waiting_for: názvy signálů nebo ID pracovních postupů, které aktuálně blokují běhcurrent_step_name: nakonfigurovaný název aktuálního kroku, pokud je k dispoziciwait_mode: aktuální režim čekání (all,anyneboquorum) při blokováníwait_details: další údaje pro kontrolu, například odpovídající a zbývající blokující položkydefinition_version: hodnota nastavená pomocíversion(), pokud existujedefinition_hash: deterministický hash definice pracovního postupuidempotency_key: trvalý klíč odeslání, pokud existujebudget: nakonfigurované limity a aktuální využití, jsou-li povolena ochranná omezeníartifact_count: počet uložených artefaktů pracovního postupuartifact_keys: klíče uložených artefaktů pro rychlou kontrolu
Queuety::retry_workflow( $workflow_id )
Opakuje neúspěšný pracovní postup od kroku, ve kterém selhal.
Queuety::retry_workflow( $workflow_id );
Vyvolá RuntimeException, pokud už byl pracovní postup po selhání kompenzován.
Queuety::pause_workflow( $workflow_id )
Pozastaví běžící pracovní postup. Již zařazené úlohy aktuálního kroku se mohou dokončit, ale další kroky se nezařadí do fronty, dokud pracovní postup neobnovíte.
Queuety::pause_workflow( $workflow_id );
Queuety::resume_workflow( $workflow_id )
Obnoví pozastavený pracovní postup.
Queuety::resume_workflow( $workflow_id );
Queuety::cancel_workflow( $workflow_id )
Zruší pracovní postup a spustí všechny zaregistrované obslužné rutiny pro úklid. Nastaví stav na cancelled a přesune všechny čekající a zpracovávané úlohy tohoto pracovního postupu do úložiště neúspěšných úloh.
Queuety::cancel_workflow( $workflow_id );
Vyvolá RuntimeException, pokud už je pracovní postup ve stavu completed nebo cancelled. Podrobnosti najdete v části Zrušení.
Queuety::rewind_workflow( $workflow_id, $to_step )
Vrátí pracovní postup k předchozímu kroku a znovu jej odtud spustí. Obnoví stav ze záznamu událostí state_after pro daný krok a zařadí do fronty následující krok.
Queuety::rewind_workflow( $workflow_id, to_step: 1 );
Vyvolá RuntimeException, pokud pro cílový krok neexistuje zaznamenaný stav. Vyžaduje protokol událostí pracovního postupu. Podrobnosti najdete v části Návrat v čase.
Queuety::fork_workflow( $workflow_id )
Vytvoří z běžícího pracovního postupu nezávislou kopii v jeho aktuálním stavu. Vrací nové ID pracovního postupu.
$forked_id = Queuety::fork_workflow( $workflow_id );
Rozvětvený pracovní postup má vlastní ID a běží nezávisle na původním. Podrobnosti najdete v části Rozvětvení.
Queuety::export_workflow( $workflow_id )
Exportuje úplnou historii provádění pracovního postupu jako pole serializovatelné do JSON. Zahrnuje řádek pracovního postupu, všechny úlohy, události pracovního postupu a položky protokolu.
$data = Queuety::export_workflow( $workflow_id );
Strukturu JSON a příklady použití najdete v části Export a přehrání.
Queuety::replay_workflow( $data )
Přehraje exportovaný pracovní postup v aktuálním prostředí. Z exportovaných dat vytvoří nový pracovní postup a vrátí jeho ID.
$new_id = Queuety::replay_workflow( $data );
Podrobnosti najdete v části Export a přehrání.
Queuety::replace_wp_cron()
Nahradí vestavěný systém WordPress cron plánovačem Queuety. Zachytává wp_schedule_event(), wp_unschedule_event() a wp_get_scheduled_event() a směruje opakující se události přes pracovní proces Queuety.
Queuety::replace_wp_cron();
Funguje pouze po načtení WordPress. Podrobnosti najdete v části Náhrada WP-Cron.
Queuety::restore_wp_cron()
Obnoví výchozí chování WordPress cron odebráním přemosťovacích háčků Queuety cron.
Queuety::restore_wp_cron();
Omezení najdete v části Náhrada WP-Cron.
Queuety::signal( $workflow_id, $name, $data )
Odešle signál pracovnímu postupu. Pokud pracovní postup na tento signál právě čeká, okamžitě se obnoví. Jinak se signál uloží a zpracuje, až pracovní postup dosáhne odpovídajícího kroku wait_for_signal.
Queuety::signal( $workflow_id, 'approved' );
// With data merged into workflow state
Queuety::signal( $workflow_id, 'approved', [
'approved_by' => 'admin@example.com',
] );
Podrobnosti najdete v části Signály pracovního postupu.
Queuety::approve_workflow( $workflow_id, $data = [], $signal_name = 'approval' )
Sémantické obalení signal() pro procesy schvalování.
Queuety::approve_workflow( $workflow_id, [
'reviewer' => 'editor@example.com',
] );
Queuety::reject_workflow( $workflow_id, $data = [], $signal_name = 'rejected' )
Sémantické obalení signal() pro procesy zamítnutí.
Queuety::reject_workflow( $workflow_id, [
'reason' => 'needs citations',
] );
Queuety::submit_workflow_input( $workflow_id, $data = [], $signal_name = 'input' )
Sémantické obalení signal() pro strukturovaný lidský vstup.
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 = [] )
Uloží jeden trvalý artefakt pracovního postupu.
Queuety::put_artifact(
$workflow_id,
'research_brief',
[ 'summary' => 'done' ],
'json',
metadata: [ 'source' => 'agent' ],
);
Queuety::put_current_artifact( $artifact_key, $content, $kind = 'json', $metadata = [] )
Uloží jeden artefakt aktuálně prováděného kroku pracovního postupu.
Queuety::put_current_artifact(
'draft',
[ 'status' => 'ready' ],
'json',
);
Queuety::workflow_artifact( $workflow_id, $artifact_key )
Načte jeden uložený artefakt pracovního postupu.
$artifact = Queuety::workflow_artifact( $workflow_id, 'draft' );
Queuety::workflow_artifacts( $workflow_id, $include_content = false )
Vypíše artefakty pracovního postupu.
$artifacts = Queuety::workflow_artifacts( $workflow_id );
Queuety::delete_workflow_artifact( $workflow_id, $artifact_key )
Odstraní jeden artefakt pracovního postupu.
Queuety::delete_workflow_artifact( $workflow_id, 'draft' );
Stavové automaty
Queuety::machine( $name )
Začne vytvářet stavový automat. Vrací StateMachineBuilder.
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 ] );
Metody StateMachineBuilder
| Metoda | Popis |
|---|---|
initial( string $state_name ) |
Výslovně nastaví počáteční stav. |
state( string $state_name, ?StateMachineStatus $terminal_status = null ) |
Zahájí konfiguraci jednoho stavu. |
action( string|array $action_class ) |
Zaregistruje zařazenou vstupní akci pro aktuální stav. Předejte název třídy nebo strukturovanou definici s class, payload a config; payload předá StateAction::handle() jako $payload. |
on( string $event, string $target, string|array|null $guard_class = null, ?string $name = null ) |
Zaregistruje jeden přechod události z aktuálního stavu. Strážní podmínka přijímá stejný strukturovaný formát jako action(). |
on_queue( string $queue ) |
Nastaví frontu používanou pro zařazené akce stavu. |
with_priority( Priority $priority ) |
Nastaví prioritu zařazených akcí stavu. |
max_attempts( int $max_attempts ) |
Nastaví limit opakování zařazených akcí stavu. |
version( string $version ) |
Označí definici automatu verzí na úrovni aplikace. |
idempotency_key( string $key ) |
Zajistí trvalé odesílání pomocí klíče zadaného volajícím. Opětovné odeslání se stejným klíčem vrátí původní ID automatu. |
dispatch( array $initial_state = [] ) |
Odešle automat a vrátí jeho ID. |
Třídy vstupních akcí stavu mohou také zpřístupňovat volitelnou metodu config() se stejnými klíči zásad pro prostředky, jaké podporují kroky pracovního postupu: concurrency_group, concurrency_limit a cost_units.
Queuety::dispatch_state_machine_definition( $definition, $initial_state = [], $options = [] )
Odešle balíček definice stavového automatu v serializované podobě. Většina kódu aplikace by měla upřednostnit machine()->dispatch(). Tato nízkoúrovňová metoda umožňuje adaptérům spustit automat bez přímé závislosti na nástroji pro jeho sestavení.
$machine_id = Queuety::dispatch_state_machine_definition(
$definition,
[ 'thread_id' => 42 ],
[ 'idempotency_key' => 'thread:42' ],
);
Queuety::machine_status( $machine_id )
Získá aktuální stav automatu. Vrací objekt StateMachineState nebo null.
$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 zpřístupňuje také:
definition_versiondefinition_hashidempotency_keyerror_messagecurrent_actionterminal_status
Queuety::machine_event( $machine_id, $event_name, $payload = [] )
Odešle externí událost čekajícímu automatu.
Queuety::machine_event(
$machine_id,
'approve',
[
'approved' => true,
'reviewer' => 'editor@example.com',
]
);
Data události jsou k dispozici strážním podmínkám přechodů a po přijetí události se sloučí s veřejným stavem automatu.
Queuety::list_machines( $limit = 50, $status = null )
Vypíše uložené záznamy automatů ke kontrole.
$machines = Queuety::list_machines();
$waiting = Queuety::list_machines( status: 'waiting_event' );
Každý záznam obsahuje ID automatu, název, stav, aktuální stav automatu, verzi definice, hash definice, idempotentní klíč a časová razítka životního cyklu.
Queuety::machine_timeline( $machine_id, $limit = 100, $offset = 0 )
Získá trvalou časovou osu událostí jednoho automatu.
$timeline = Queuety::machine_timeline( $machine_id );
$page_two = Queuety::machine_timeline( $machine_id, 100, 100 );
Každý řádek události obsahuje: 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 a created_at.
Queuety::machine_trace( $machine_id )
Získá normalizovaný balíček trasování pro ladicí nástroj stavového automatu UI.
$trace = Queuety::machine_trace( $machine_id );
Trasování obsahuje machine, seskupené states, nezpracované events, korelovanou akci jobs, logs, artifacts a chunks.
Queuety::machine_state_at( $machine_id, $event_id )
Získá veřejný stav po konkrétní události trasování automatu, případně null, pokud nebyl nalezen.
$state = Queuety::machine_state_at( $machine_id, $event_id );
Model životního cyklu a vzorce použití najdete v části Stavové automaty.
Šablony pracovních postupů
Queuety::define_workflow( $name )
Vytvoří nástroj pro sestavení pracovního postupu, který lze použít jako šablonu.
$builder = Queuety::define_workflow( 'onboarding' )
->then( CreateAccountHandler::class )
->then( SendWelcomeEmailHandler::class );
Queuety::register_workflow_template( $builder )
Zaregistruje šablonu pracovního postupu.
Queuety::register_workflow_template( $builder );
Queuety::run_workflow( $name, $payload, $options = [] )
Odešle zaregistrovanou šablonu pracovního postupu podle názvu. Vrací ID pracovního postupu.
$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' ],
);
Vyvolá RuntimeException, pokud šablona není zaregistrována.
Queuety::dispatch_workflow_definition( $definition, $payload, $options = [] )
Odešle pracovní postup z balíčku definice za běhu.
Většina kódu aplikace by měla upřednostnit run_workflow() nebo workflow()->dispatch(). Tato nízkoúrovňová metoda umožňuje opakovaně použitelné adaptéry odeslat serializovanou definici pracovního postupu bez závislosti na interních metodách správce pracovních postupů.
$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 )
Zaregistruje akci WordPress, která odešle pracovní postup.
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 může být:
- název zaregistrované šablony pracovního postupu
- vložené
WorkflowBuilder
map, when a idempotency_key se spouštějí synchronně v rámci aktuálního požadavku WordPress. Po odeslání se asynchronně spouštějí pouze samotné kroky pracovního postupu.
Úplného průvodce najdete v části Spouštěče akcí.
Queuety::workflow_templates()
Získá registr šablon pracovních postupů.
$registry = Queuety::workflow_templates();
$template = $registry->get( 'onboarding' ); // WorkflowTemplate or null
Queuety::load_workflows( $directory, $recursive )
Načte a zaregistruje všechny soubory pracovních postupů z adresáře. Každý soubor .php by měl vracet WorkflowBuilder. Vrací počet zaregistrovaných pracovních postupů.
$count = Queuety::load_workflows( __DIR__ . '/workflows/' );
$count = Queuety::load_workflows( __DIR__ . '/workflows/', recursive: true );
Queuety::load_workflow_file( $file_path )
Načte a zaregistruje jeden soubor pracovního postupu. Vrací WorkflowTemplate nebo null, pokud soubor nevrací nástroj pro sestavení.
$template = Queuety::load_workflow_file( __DIR__ . '/workflows/onboard-user.php' );
Plánování
Queuety::schedule( $handler, $payload )
Vytvoří opakovaný plán. Vrací nástroj pro sestavení PendingSchedule.
Queuety::schedule( 'cleanup_handler' )
->every( '1 hour' )
->on_queue( 'maintenance' );
Queuety::schedule( 'nightly_report' )
->cron( '0 3 * * *' )
->on_queue( 'reports' );
Metody PendingSchedule
| Metoda | Popis |
|---|---|
every( string $interval ) |
Nastaví výraz intervalu (e.g. 1 hour, 30 minutes) |
cron( string $expression ) |
Nastaví výraz cron s 5 poli |
overlap( OverlapPolicy $policy ) |
Nastaví zásadu překrývání (výchozí: Allow). Viz Plánování. |
on_queue( string $queue ) |
Nastaví cílovou frontu (výchozí: default) |
id() |
Vynutí odeslání a vrátí ID plánu |
Queuety::scheduler()
Získá interní instanci Scheduler.
$scheduler = Queuety::scheduler();
$schedules = $scheduler->list();
$scheduler->remove( 'handler_name' );
$count = $scheduler->tick(); // manually trigger
Registrace obslužných tříd
Queuety::register( $name, $class )
Zaregistruje třídu obslužné rutiny pod daným názvem.
Queuety::register( 'send_email', SendEmailHandler::class );
Queuety::discover_handlers( $directory, $namespace )
Automaticky vyhledá a zaregistruje třídy obslužných rutin z adresáře. Vrátí počet zaregistrovaných obslužných rutin.
$count = Queuety::discover_handlers( __DIR__ . '/handlers', 'MyPlugin\\Handlers' );
Queuety::registry()
Získá registr obslužných rutin.
$registry = Queuety::registry();
Sledovatelnost
Queuety::logger()
Získá instanci Logger pro dotazování na záznamy protokolu.
$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()
Získá instanci Metrics.
$stats = Queuety::metrics()->handler_stats( 60 ); // last 60 minutes
Queuety::webhook_notifier()
Získá instanci WebhookNotifier.
$notifier = Queuety::webhook_notifier();
$id = $notifier->register( 'job.buried', 'https://hooks.slack.com/...' );
$webhooks = $notifier->list();
$notifier->remove( $id );
Queuety::rate_limiter()
Získá instanci RateLimiter.
$limiter = Queuety::rate_limiter();
$limiter->register( 'call_openai', 60, 60 );
Mezipaměť
Queuety::cache()
Získá instanci aktivního backendu mezipaměti.
$cache = Queuety::cache();
$cache->set( 'my_key', 'my_value', 60 );
$value = $cache->get( 'my_key' );
Queuety::set_cache( $cache )
Přepíše backend mezipaměti. Chcete-li použít vlastní implementaci, zavolejte Queuety::init() předem.
Queuety::set_cache( new RedisCache() );
Queuety::init( $connection );
QUEUETY_CACHE_TTL
Výchozí TTL mezipaměti v sekundách (výchozí: 5). Chcete-li změnit výchozí TTL používané interními operacemi mezipaměti, nastavte tuto konstantu v wp-config.php.
define( 'QUEUETY_CACHE_TTL', 10 );
Podrobnosti o backendech a automatické detekci najdete v části Ukládání do mezipaměti.
Protokol událostí stavového automatu
Queuety::machine_events()
Získá instanci StateMachineEventLog.
$log = Queuety::machine_events();
Protokol událostí workflow
Queuety::workflow_events()
Získá instanci WorkflowEventLog.
$log = Queuety::workflow_events();
Queuety::workflow_timeline( $workflow_id, $limit = 100, $offset = 0 )
Získá úplnou časovou osu událostí workflow. Vrátí pole řádků událostí seřazených podle ID.
$events = Queuety::workflow_timeline( $workflow_id );
$next = Queuety::workflow_timeline( $workflow_id, 100, 100 );
Každý řádek události obsahuje: 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 (zkopírováno z error.message), duration_ms a created_at.
Queuety::workflow_trace( $workflow_id )
Získá normalizovaný soubor tras pro ladicí nástroj workflow UI.
$trace = Queuety::workflow_trace( $workflow_id );
Trasování obsahuje workflow, seskupené steps, nezpracované events, jobs, logs, artifacts, chunks, signals a wait_dependencies.
Queuety::trace_input( $input )
Nastaví vstup na úrovni aplikace pro právě prováděný krok workflow.
Queuety::trace_output( $output )
Nastaví výstup na úrovni aplikace pro právě prováděný krok workflow.
Queuety::trace_context( array $context )
Sloučí kontext na úrovni aplikace s právě prováděným krokem workflow.
Queuety::workflow_state_at( $workflow_id, $step_index )
Získá veřejný stav po dokončení nebo obnovení konkrétního kroku workflow, případně null, pokud nebyl nalezen.
$state = Queuety::workflow_state_at( $workflow_id, 2 );
Příklady použití a příkazy CLI najdete v části Protokol událostí workflow.
Interní instance
Tyto metody vracejí interní instance pro pokročilé případy použití.
| Metoda | Vrací | Popis |
|---|---|---|
Queuety::queue() |
Queue |
Operace s frontou (převzetí, uvolnění, odložení) |
Queuety::worker() |
Worker |
Řízení pracovních procesů |
Queuety::workflow_manager() |
Workflow |
Orchestrace workflow |
Queuety::machines() |
StateMachine |
Správa životního cyklu stavového automatu |
Queuety::connection() |
Connection |
Připojení k databázi |
Queuety::batch_manager() |
BatchManager |
Správa životního cyklu dávky (zrušení, odstranění) |
Queuety::chunk_store() |
ChunkStore |
Trvalé ukládání částí pro streamovací kroky |
Queuety::machine_events() |
StateMachineEventLog |
Události trasování stavového automatu a normalizované soubory tras |
Queuety::workflow_events() |
WorkflowEventLog |
Události trasování workflow a normalizované soubory tras |
Queuety::resource_manager() |
ResourceManager |
Profily přijímání pracovníků, vážené rozpočty a kontrola systémové paměti |
Queuety::cache() |
Cache |
Instance backendu mezipaměti |
Testování
Queuety::fake()
Pro testování nahraďte frontu falešnou frontou v paměti. Vrátí instanci QueueFake, která zaznamenává odeslané úlohy, aniž by přistupovala k databázi. create_batch() zaznamenává také falešná metadata dávek; řetězce se ověřují prostřednictvím jednotlivých odeslaných úloh.
$fake = Queuety::fake();
SendEmailJob::dispatch( 'user@example.com', 'Hello', 'Hi!' );
$fake->assert_pushed( SendEmailJob::class );
Queuety::reset()
Resetuje veškerý interní stav včetně falešné fronty. Volejte v tearDown().
Queuety::reset();
Metody QueueFake
| Metoda | Popis |
|---|---|
assert_pushed( string $class, ?Closure $callback ) |
Ověří, že byla zařazena třída úlohy. Volitelná zpětná vazba filtruje podle dat úlohy. |
assert_pushed_times( string $class, int $count ) |
Ověří, že byla úloha zařazena přesně Nkrát. |
assert_not_pushed( string $class ) |
Ověří, že úloha nebyla nikdy zařazena. |
assert_nothing_pushed() |
Ověří, že nebyla zařazena žádná úloha. |
assert_batched( ?Closure $callback ) |
Ověří, že byla odeslána dávka. Volitelná zpětná vazba filtruje podle dat dávky. |
pushed( string $class ) |
Získá všechna zaznamenaná odeslání pro třídu. Vrátí pole. |
batches() |
Získá všechny zaznamenané dávky. Vrátí pole. |
reset() |
Vymaže všechny zaznamenané úlohy a dávky. |
Úplné příklady najdete v části Testování.
Rozhraní
Handler
Rozhraní pro jednoduché obslužné rutiny úloh.
interface Handler {
public function handle( array $payload ): void;
public function config(): array;
}
config() může vrátit queue, max_attempts, backoff, rate_limit, concurrency_group, concurrency_limit a cost_units.
Step
Rozhraní pro obslužné rutiny kroků workflow.
interface Step {
public function handle( array $state ): array;
public function config(): array;
}
config() může vrátit max_attempts, backoff, concurrency_group, concurrency_limit a cost_units.
Contracts\ForEachHandler
Rozhraní pro obslužné rutiny dynamických větví for-each.
namespace Queuety\Contracts;
interface ForEachHandler {
public function handle_item( array $state, mixed $item, int $index ): array;
public function config(): array;
}
config() může vrátit max_attempts, backoff, rate_limit, concurrency_group, concurrency_limit a cost_units.
Contracts\ForEachReducer
Volitelná redukční funkce pro výsledky for-each.
namespace Queuety\Contracts;
interface ForEachReducer {
public function reduce( array $state, array $for_each ): array;
}
Contracts\Compensation
Rozhraní pro obslužné rutiny kompenzace kroků.
namespace Queuety\Contracts;
interface Compensation {
public function handle( array $state ): void;
}
Contracts\RepeatCondition
Rozhraní pro predikáty smyčky používané funkcemi repeat_until() a repeat_while(), pokud je porovnání klíčů stavu příliš omezené.
namespace Queuety\Contracts;
interface RepeatCondition {
public function matches( array $state ): bool;
}
Contracts\StateAction
Rozhraní pro akce zařazené při vstupu do stavu.
namespace Queuety\Contracts;
interface StateAction {
public function handle( array $state, ?string $event = null, array $event_payload = [], array $payload = [] ): array|string;
}
$payload přebírá strukturované payload z definice akce, pokud byl automat definován s ním. Vrací aktualizace veřejného stavu. Pokud má akce okamžitě přejít do dalšího stavu, uveďte _event a případně _event_payload.
Akce stavů mohou také definovat volitelnou metodu config(). Když je akce při vstupu zařazena do fronty, Queuety z ní načte concurrency_group, concurrency_limit a cost_units.
Contracts\StateGuard
Rozhraní pro ochranné podmínky přechodů stavového automatu.
namespace Queuety\Contracts;
interface StateGuard {
public function allows( array $state, array $event_payload, string $event, array $payload = [] ): bool;
}
Contracts\Job
Rozhraní pro samostatné třídy úloh, které lze odeslat. Třídy úloh zapouzdřují svou datovou část jako veřejné vlastnosti a logiku provádění v handle().
namespace Queuety\Contracts;
interface Job {
public function handle(): void;
}
Třídy úloh mohou volitelně definovat metodu middleware(), která vrací pole instancí Contracts\Middleware.
Contracts\Cache
Rozhraní pro backendy mezipaměti. Vestavěné implementace a vlastní backendy najdete v části Ukládání do mezipaměti.
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;
}
| Metoda | Vrací | Popis |
|---|---|---|
get( $key ) |
mixed |
Načte hodnotu. Pokud nebyla nalezena nebo její platnost vypršela, vrátí null. |
set( $key, $value, $ttl ) |
void |
Uloží hodnotu. TTL 0 znamená, že platnost nevyprší. |
delete( $key ) |
void |
Odebere hodnotu. |
has( $key ) |
bool |
Ověří, zda klíč existuje a nevypršela jeho platnost. |
add( $key, $value, $ttl ) |
bool |
Atomicky nastaví hodnotu, jen pokud dosud neexistuje. Vrátí true, pokud byla hodnota nastavena. |
flush() |
void |
Odebere z mezipaměti všechny položky. |
WorkflowEventLog
Zaznamenává přechody kroků workflow včetně úplného kontextu trasování. Přístup přes Queuety::workflow_events().
| Metoda | Vrací | Popis |
|---|---|---|
record_event( array $event ) |
void |
Zaznamená nezpracovanou událost trasování |
record_step_started( ... ) |
void |
Zaznamená událost step_started se vstupem a stavem před spuštěním |
record_step_completed( ... ) |
void |
Zaznamená událost step_completed se vstupem, výstupem a stavem před spuštěním i po něm |
record_step_failed( ... ) |
void |
Zaznamená událost step_failed s kontextem trasování a strukturovanými podrobnostmi o chybě |
get_timeline( int $workflow_id, ?int $limit = 100, int $offset = 0 ) |
array |
Získá události trasování workflow v pořadí, případně po stránkách |
get_trace( int $workflow_id ) |
array |
Získá normalizovaný soubor tras |
get_state_at_step( int $workflow_id, int $step_index ) |
array|null |
Získá veřejný stav po dokončení nebo obnovení konkrétního kroku |
prune( int $older_than_days ) |
int |
Odstraní události starší než zadaný počet dní. Vrátí počet odstraněných řádků. |
StateMachineEventLog
Zaznamenává události životního cyklu stavového automatu, akcí, přechodů a ochranných podmínek se stejnými poli trasování jako události workflow.
| Metoda | Vrací | Popis |
|---|---|---|
record_event( array $event ) |
void |
Zaznamená nezpracovanou událost trasování automatu |
get_timeline( int $machine_id, ?int $limit = 100, int $offset = 0 ) |
array |
Získá události trasování automatu v pořadí, případně po stránkách |
get_trace( int $machine_id ) |
array |
Získá normalizovaný soubor trasování automatu |
get_state_at_event( int $machine_id, int $event_id ) |
array|null |
Získá veřejný stav po konkrétní události trasování automatu |
prune( int $older_than_days ) |
int |
Odstraní události starší než zadaný počet dní. Vrátí počet odstraněných řádků. |
Contracts\StreamingStep
Rozhraní pro obslužné rutiny kroků workflow, které vytvářejí proud částí. Každá předaná hodnota se okamžitě uloží do databáze. Viz Streamovací kroky.
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;
}
| Metoda | Popis |
|---|---|
stream( $state, $existing_chunks ) |
Generátor, který předává textové části. Po každém předání se okamžitě zapíše do DB. Při opakování pokusu obsahuje $existing_chunks již uložené části. |
on_complete( $chunks, $state ) |
Zavolá se po dokončení proudu. Vrátí data ke sloučení se stavem workflow. |
config() |
Volitelná konfigurace. Podporuje max_attempts, backoff, concurrency_group, concurrency_limit a cost_units. |
ChunkStore
Spravuje trvalé ukládání částí pro streamovací kroky. Přístup přes Queuety::chunk_store().
| Metoda | Vrací | Popis |
|---|---|---|
get_chunks( int $job_id ) |
string[] |
Načte všechny části úlohy seřazené podle indexu části |
append_chunk( int $job_id, int $chunk_index, string $content, ?int $workflow_id, ?int $step_index ) |
void |
Připojí jednu část |
clear_chunks( int $job_id ) |
void |
Odstraní všechny části úlohy |
chunk_count( int $job_id ) |
int |
Spočítá části úlohy |
get_accumulated( int $job_id ) |
string |
Spojí všechny části do jednoho řetězce |
Heartbeat
Statický pomocník pro odesílání signálů aktivity z obslužných rutin kroků a streamovacích kroků. Viz Signály aktivity.
| Metoda | Popis |
|---|---|
Heartbeat::init( int $job_id, Connection $conn ) |
Nastaví kontext úlohy. Pracovník ji volá automaticky. |
Heartbeat::beat( array $progress = [] ) |
Odešle signál aktivity. Aktualizuje reserved_at a volitelně uloží údaje o průběhu do heartbeat_data. |
Heartbeat::clear() |
Vymaže kontext úlohy. Pracovník ji volá automaticky. |
Heartbeat::current_job_id() |
Získá aktuální ID úlohy (pro testování). |
Contracts\Middleware
Rozhraní pro middleware úloh. Middleware obaluje provádění úlohy do vícevrstvého řetězce.
namespace Queuety\Contracts;
interface Middleware {
public function handle( object $job, \Closure $next ): void;
}
Traity
Dispatchable
Poskytuje statické metody pro odesílání u tříd úloh implementujících Contracts\Job. Argumenty konstruktoru se předávají při vytváření instance úlohy.
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' );
| Metoda | Vrací | Popis |
|---|---|---|
dispatch( ...$args ) |
PendingJob |
Vytvoří instanci a odešle ji |
dispatch_if( bool $condition, ...$args ) |
PendingJob|null |
Odešle pouze tehdy, když je $condition true |
dispatch_unless( bool $condition, ...$args ) |
PendingJob|null |
Odešle pouze tehdy, když je $condition false |
with_chain( array $jobs ) |
ChainBuilder |
Zahájí řetězec z této třídy úlohy |
// 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();
Vlastnosti úlohy
Třídy úloh mohou deklarovat veřejné vlastnosti, které pracovník načítá pomocí reflexe a jimiž lze přepsat chování při opakování pokusů a časových limitech.
| Vlastnost | Typ | Popis |
|---|---|---|
$tries |
int |
Maximální počet pokusů, než bude úloha odložena |
$timeout |
int |
Maximální doba provádění v sekundách (vyžaduje pcntl) |
$backoff |
array |
Postupně se prodlužující prodlevy v sekundách pro jednotlivé pokusy |
class ImportProductsJob implements Job {
use Dispatchable;
public int $tries = 5;
public int $timeout = 120;
public array $backoff = [ 10, 30, 60, 120 ];
// ...
}
Háček failed()
Když je úloha po vyčerpání všech opakovaných pokusů odložena, pracovník zavolá metodu failed() instance úlohy, pokud je definována.
public function failed( \Throwable $exception ): void {
error_log( "Job failed: {$exception->getMessage()}" );
}
Vestavěný middleware
Všechny třídy middleware implementují Contracts\Middleware a nacházejí se v jmenném prostoru Queuety\Middleware.
| Třída | Popis |
|---|---|
RateLimited( int $max, int $window ) |
Vynutí limit $max spuštění za $window sekund |
Timeout( int $seconds ) |
Ukončí úlohu po $seconds sekundách (vyžaduje pcntl) |
UniqueJob( ?string $key ) |
Zabrání souběžnému spuštění pomocí databázového zámku |
WithoutOverlapping( string $key, int $release_after ) |
Zabrání překrývajícímu se spuštění s automatickým vypršením zámku |
ThrottlesExceptions( int $max_attempts, int $decay_minutes ) |
Omezuje výjimky, aby zabránila zahlcení úlohami při výpadku externí služby |
Podrobnosti o použití a příklady najdete v části Middleware.
Výčtové typy
Priority
enum Priority: int {
case Low = 0;
case Normal = 1;
case High = 2;
case Urgent = 3;
}
JobStatus
enum JobStatus: string {
case Pending = 'pending';
case Processing = 'processing';
case Completed = 'completed';
case Failed = 'failed';
case Buried = 'buried';
}
WorkflowStatus
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
enum StateMachineStatus: string {
case Running = 'running';
case WaitingEvent = 'waiting_event';
case Completed = 'completed';
case Failed = 'failed';
case Paused = 'paused';
case Cancelled = 'cancelled';
}
BackoffStrategy
enum BackoffStrategy: string {
case Fixed = 'fixed';
case Linear = 'linear';
case Exponential = 'exponential';
}
ForEachMode
enum ForEachMode: string {
case All = 'all';
case FirstSuccess = 'first_success';
case Quorum = 'quorum';
}
WaitMode
enum WaitMode: string {
case All = 'all';
case Any = 'any';
case Quorum = 'quorum';
}
LogEvent
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
enum OverlapPolicy: string {
case Allow = 'allow';
case Skip = 'skip';
case Buffer = 'buffer';
}
Podrobnosti o fungování jednotlivých zásad najdete v části Plánování.
Atributy
#[QueuetyHandler]
Atribut PHP 8 pro automatickou registraci. Viz Atributy PHP.
use Queuety\Attributes\QueuetyHandler;
#[QueuetyHandler( name: 'send_email', queue: 'emails', max_attempts: 5 )]
class SendEmailHandler implements Handler {
// ...
}
Most CLI
Queuety::cli_command()
Vrátí zveřejněný název kořenového příkazu WP-CLI.
Výchozí kořenový příkaz je queuety. Chcete-li příkaz zveřejnit pod jiným názvem, definujte QUEUETY_CLI_COMMAND před spuštěním Queuety.
Queuety::cli_command_map()
Vrátí serializovatelný katalog příkazů Queuety CLI pro testovací nástroje, které potřebují znát dostupné příkazy bez procházení tříd WP-CLI.
Každá položka obsahuje:
- stabilní ID operace
- relativní cestu CLI
- třídu a metodu obslužné rutiny WP-CLI
- možné cíle spuštění
- adaptér, který normalizuje nezpracované argumenty CLI na plán spuštění
Queuety::resolve_cli_command( $path, $args, $assoc_args )
Převede analyzované volání CLI na plán spuštění, který má testovací nástroj provést.
$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'] );
Plány spuštění zohledňují přenosový protokol:
- příkazy se překládají na veřejně dostupné funkce PHP s normalizovanými argumenty
- katalog předem uvádí všechny možné cíle spuštění, zatímco
callableatransporturčují větev vybranou pro analyzovaný vstup
Toto je standardní Queuety používaný pro testovací nástroje agentů: stabilní ID operace, veřejný cíl API a normalizace spravovaná adaptérem pro záležitosti specifické pro CLI, jako jsou výchozí fronty, dekódování datové části JSON, větvení fondu pracovníků a zpracování souborů exportu workflow.