PHP API

Vizualizați ca Markdown Editați pe GitHub Documentație la 2cc7b2b

Tot ce ține de Queuety este accesibil prin clasa statică Queuety\Queuety. Aceasta este punctul unic de intrare pentru distribuirea sarcinilor, construirea fluxurilor de lucru, operarea mașinilor de stări, gestionarea cozilor și accesarea subsistemelor interne.

PHP

use Queuety\Queuety;

Acces autonom la nucleu

Runtime-ul Queuety, susținut de DB, poate fi inițializat în afara WordPress.

Această cale autonomă este destinată nucleului de orchestrare:

sarcini și workeri (pool-urile de workeri au nevoie și de constantele DB_HOST, DB_NAME, DB_USER și DB_PASSWORD; consultați secțiunea de mai jos)
fluxuri de lucru și semnale
mașini de stări și evenimente
artefacte, jurnale, metrici și programări

Nu înlocuiește adaptorul WordPress. Queuety::on_action(), replace_wp_cron(), inițializarea pluginului și calea reală de înregistrare a wp queuety ... necesită în continuare WordPress.

`Queuety::init( Connection $conn )`

Inițializați Queuety cu o conexiune directă susținută de 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 );

Argumentul opțional table_prefix modifică partea Queuety a numelor tabelelor runtime pentru fiecare conexiune:

PHP

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

Astfel, tabele precum queuety_jobs se rezolvă la wp_themequeue_jobs.

`Queuety::ensure_schema()`

Asigurați-vă că tabelele runtime Queuety există pentru conexiunea activă.

PHP

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

Pentru modelul complet și limita WordPress, consultați Utilizare autonomă.

Sarcini

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

Distribuie o sarcină. Acceptă fie un șir cu numele unui handler, fie o instanță Contracts\Job. Returnează un builder PendingJob pentru înlănțuirea opțiunilor.

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

Când se transmite o instanță Contracts\Job, serializatorul extrage FQCN-ul ca nume al handlerului și proprietățile publice ca payload. Parametrul $payload este ignorat când $handler este o instanță Job.

Metode `PendingJob`

Metodă	Descriere
`on_queue( string $queue )`	Setează coada țintă
`with_priority( Priority $priority )`	Setează nivelul de prioritate
`delay( int $seconds )`	Întârziere până când sarcina devine disponibilă
`max_attempts( int $max )`	Numărul maxim de încercări repetate
`rate_limit( int $max, int $window )`	Limită de rată: numărul maxim de execuții per fereastră, în secunde
`unique()`	Împiedică distribuirile duplicate
`after( int $job_id )`	Așteaptă mai întâi finalizarea altei sarcini
`concurrency_group( string $group, int $limit )`	Plasează sarcina într-un grup comun de admitere a workerilor
`cost_units( int $units )`	Atașează la distribuire metadate despre costul relativ de execuție
`id()`	Forțează distribuirea și returnează ID-ul sarcinii

concurrency_group() și cost_units() contribuie la modelul mai amplu Gestionarea resurselor, astfel încât aceeași distribuire poate participa la limite de concurență bazate pe număr, bugete ponderate pentru cozi sau provideri și euristici de admitere a workerilor.

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

Execută sincron o sarcină în procesul curent, fără a o adăuga în coadă.

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' ] );

Metoda handle() a sarcinii este apelată direct. Nu se creează niciun rând în baza de date, nu este implicat niciun worker și nu se aplică middleware.

`Queuety::batch( $jobs )`

Distribuie mai multe sarcini printr-o singură instrucțiune INSERT cu mai multe rânduri.

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

Creează un BatchBuilder pentru distribuirea unui grup de sarcini cu callback-uri și urmărirea progresului. Consultați Grupuri de sarcini pentru detalii complete.

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

Metode `BatchBuilder`

Metodă	Returnează	Descriere
`name( string $name )`	`self`	Setează un nume pentru grupul de sarcini
`on_queue( string $queue )`	`self`	Setează coada pentru toate sarcinile din grup
`then( string $handler_class )`	`self`	Handler apelat când grupul se finalizează cu succes
`catch( string $handler_class )`	`self`	Handler apelat când prima sarcină din grup eșuează
`finally( string $handler_class )`	`self`	Handler apelat când grupul se încheie (cu succes sau eșec)
`allow_failures()`	`self`	Permite callback-ului `then` să se declanșeze chiar și în caz de eșecuri
`dispatch()`	`Batch`	Creează rândul grupului și toate sarcinile; returnează obiectul valoare `Batch`

`Queuety::find_batch( $id )`

Găsește un grup de sarcini după ID. Returnează un obiect valoare Batch sau null.

PHP

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

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

Controlul workerilor

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

Rulează direct o buclă de worker.

PHP

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

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

Rulează direct un pool de workeri cu dimensiune fixă. Necesită extensiile pcntl și posix. Pool-ul citește constantele WordPress DB_HOST, DB_NAME, DB_USER și DB_PASSWORD pentru conexiunile proceselor copil (doar prefixele tabelelor provin din conexiunea inițializată), așadar definiți-le când rulați un pool în afara WordPress (#5).

PHP

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

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

Rulează direct un pool adaptiv de workeri. Aceleași cerințe ca pentru run_worker_pool().

PHP

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

Obiectul valoare `Batch`

Metodă/Proprietate	Tip	Descriere
`$id`	`int`	ID-ul grupului de sarcini
`$name`	`string\|null`	Numele opțional al grupului
`$total_jobs`	`int`	Numărul total de sarcini din grup
`$pending_jobs`	`int`	Sarcini încă în așteptare
`$failed_jobs`	`int`	Numărul de sarcini eșuate
`$failed_job_ids`	`array`	ID-urile sarcinilor eșuate
`$created_at`	`DateTimeImmutable`	Momentul creării grupului
`$finished_at`	`DateTimeImmutable\|null`	Momentul încheierii grupului
`$cancelled_at`	`DateTimeImmutable\|null`	Momentul anulării grupului
`progress()`	`int`	Procentul de finalizare (0-100)
`finished()`	`bool`	Dacă grupul s-a încheiat
`has_failures()`	`bool`	Dacă vreo sarcină a eșuat
`cancelled()`	`bool`	Dacă grupul a fost anulat

`Queuety::chain( $jobs )`

Creează un ChainBuilder pentru executarea secvențială a sarcinilor. Fiecare sarcină depinde de finalizarea cu succes a celei precedente.

PHP

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

Metode `ChainBuilder`

Metodă	Returnează	Descriere
`on_queue( string $queue )`	`self`	Setează coada pentru toate sarcinile din lanț
`catch( string $handler_class )`	`self`	Handler apelat când orice sarcină din lanț este îngropată definitiv
`dispatch()`	`int`	Creează toate sarcinile cu dependențe secvențiale; returnează ID-ul primei sarcini

`Queuety::retry( $job_id )`

Reîncearcă o anumită sarcină. O resetează la starea pending.

PHP

Queuety::retry( 42 );

`Queuety::retry_buried()`

Reîncearcă toate sarcinile îngropate. Returnează numărul sarcinilor reîncercate.

PHP

$count = Queuety::retry_buried();

`Queuety::purge( $older_than_days )`

Elimină sarcinile finalizate. Returnează numărul sarcinilor eliminate.

PHP

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

Cozi

`Queuety::stats( $queue )`

Obține numărul de sarcini grupate după stare.

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

Obține toate sarcinile îngropate.

PHP

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

`Queuety::pause( $queue )`

Pune o coadă în pauză, astfel încât workerii să o ignore.

PHP

Queuety::pause( 'emails' );

`Queuety::resume( $queue )`

Reia o coadă aflată în pauză.

PHP

Queuety::resume( 'emails' );

`Queuety::is_paused( $queue )`

Verifică dacă o coadă este în pauză.

PHP

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

Fluxuri de lucru

`Queuety::workflow( $name )`

Începeți să construiți un flux de lucru. Returnează un WorkflowBuilder.

PHP

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

Metode `WorkflowBuilder`

Metodă	Descriere
`then( string $handler_class, ?string $name = null )`	Adaugă un pas secvențial. Numele este opțional și poate fi folosit pentru ramificare condițională.
`parallel( array $handler_classes )`	Adaugă un grup de pași paraleli.
`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 )`	Adaugă un pas dinamic for-each care extinde câte o sarcină pe ramură pentru fiecare element din starea fluxului de lucru.
`repeat_while( string $target_step, ?string $state_key = null, mixed $expected = true, ?string $name = null, ?string $condition_class = null, ?int $max_iterations = null )`	Revine la un pas anterior cu nume, cât timp o cheie publică de stare sau o clasă `RepeatCondition` corespunde.
`repeat_until( string $target_step, ?string $state_key = null, mixed $expected = true, ?string $name = null, ?string $condition_class = null, ?int $max_iterations = null )`	Revine la un pas anterior cu nume, până când o cheie publică de stare sau o clasă `RepeatCondition` corespunde.
`run_workflow( string $workflow_name, WorkflowBuilder $sub_builder )`	Adaugă un pas de rulare a fluxului de lucru.
`delay( int $seconds = 0, int $minutes = 0, int $hours = 0, int $days = 0 )`	Adaugă un pas de întârziere durabilă. Duratele se însumează.
`wait_for_signal( string $name, ?string $result_key = null, ?string $step_name = null, array $match_payload = [], ?string $correlation_key = null )`	Adaugă un pas de așteptare a unui semnal. Opționalele `match_payload` și `correlation_key` permit ignorarea evenimentelor fără legătură.
`wait_for_signals( array $signal_names, WaitMode $mode = WaitMode::All, ?string $result_key = null, ?string $name = null, array $match_payload = [], ?string $correlation_key = null )`	Așteaptă toate sau oricare dintre mai multe semnale, cu potrivire opțională a payload-ului și corelare.
`wait_for_approval( string $signal_name = 'approval', ?string $result_key = 'approval', ?string $name = null, array $match_payload = [], ?string $correlation_key = null )`	Așteaptă un semnal de aprobare umană.
`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 )`	Așteaptă aprobarea sau respingerea și stochează un payload structurat al deciziei.
`wait_for_input( string $signal_name = 'input', ?string $result_key = 'input', ?string $name = null, array $match_payload = [], ?string $correlation_key = null )`	Așteaptă un semnal de introducere a datelor de către o persoană.
`wait_for_workflow( int\|string $workflow, ?string $result_key = null, ?string $name = null )`	Așteaptă finalizarea unui ID de flux de lucru sau a unui ID de flux de lucru stocat în starea publică.
`wait_for_workflows( array\|string $workflows, WaitMode $mode = WaitMode::All, ?string $result_key = null, ?string $name = null, ?int $quorum = null )`	Așteaptă finalizarea tuturor, a oricăruia sau a unui cvorum de mai multe fluxuri de lucru.
`wait_for_workflow_group( string $group_key, WaitMode $mode = WaitMode::All, ?int $quorum = null, ?string $result_key = null, ?string $name = null )`	Așteaptă un grup de fluxuri de lucru pornit anterior, folosind cheia internă a grupului.
`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 )`	Distribuie un flux de lucru de nivel superior pentru fiecare element runtime, cu opțiunea de a eticheta setul pornit cu o cheie de grup durabilă.
`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 )`	Alias destinat agenților pentru `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 )`	Alias destinat agenților pentru `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 )`	Alias destinat agenților pentru `wait_for_workflow_group()`.
`version( string $version )`	Etichetează definiția fluxului de lucru cu o versiune la nivel de aplicație pentru inspectare și exporturi.
`idempotency_key( string $key )`	Face durabilă distribuirea fluxului de lucru folosind o cheie furnizată de apelant. O nouă distribuire cu aceeași cheie returnează ID-ul inițial al fluxului de lucru.
`max_transitions( int $max )`	Eșuează fluxul de lucru dacă se finalizează mai mult de `$max` tranziții de pas.
`max_for_each_items( int $max )`	Eșuează fluxul de lucru dacă un pas `for_each()` încearcă să extindă mai mult de `$max` elemente.
`max_state_bytes( int $bytes )`	Eșuează fluxul de lucru dacă starea publică depășește `$bytes`.
`max_cost_units( int $max )`	Eșuează fluxul de lucru dacă pașii finalizați consumă mai mult de `$max` unități de cost.
`max_started_workflows( int $max )`	Eșuează fluxul de lucru înainte de a distribui mai mult de `$max` fluxuri de lucru copil.
`compensate_with( string $handler_class )`	Atașează un handler de compensare la cel mai recent pas adăugat.
`compensate_on_failure()`	Rulează automat compensările pașilor finalizați când fluxul de lucru eșuează.
`on_queue( string $queue )`	Setează coada pentru toți pașii.
`with_priority( Priority $priority )`	Setează prioritatea pentru toți pașii.
`max_attempts( int $max )`	Setează numărul maxim de încercări repetate pentru fiecare pas.
`on_cancel( string $handler_class )`	Înregistrează o clasă handler de curățare pentru anulare. Consultați Anulare.
`prune_state_after( int $steps = 2 )`	Activează eliminarea automată a stării. Elimină rezultatele pașilor mai vechi decât pragul. Consultați Eliminarea stării.
`must_complete_within( int $seconds = 0, int $minutes = 0, int $hours = 0, int $days = 0 )`	Setează un termen-limită pentru fluxul de lucru. Duratele se însumează. Consultați Termene-limită.
`on_deadline( string $handler_class )`	Înregistrează un handler apelat când se depășește termenul-limită al fluxului de lucru. Consultați Termene-limită.
`dispatch( array $initial_payload = [] )`	Distribuie fluxul de lucru. Returnează ID-ul fluxului de lucru.

`Queuety::workflow_status( $workflow_id )`

Obține starea fluxului de lucru. Returnează un obiect WorkflowState sau 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 expune și:

waiting_for: numele semnalelor sau ID-urile fluxurilor de lucru care blochează în prezent execuția
current_step_name: numele configurat al pasului curent, dacă există
wait_mode: modul de așteptare curent (all, any sau quorum) cât timp execuția este blocată
wait_details: date suplimentare de inspecție, cum ar fi blocajele identificate și cele rămase
definition_version: valoarea setată de version(), dacă există
definition_hash: un hash determinist al definiției fluxului de lucru
idempotency_key: cheia durabilă de distribuire, dacă există
budget: limitele configurate și utilizarea curentă, dacă sunt activate mecanismele de protecție
artifact_count: numărul de artefacte stocate pentru fluxul de lucru
artifact_keys: cheile artefactelor stocate, pentru inspecție rapidă

`Queuety::retry_workflow( $workflow_id )`

Reia un flux de lucru eșuat de la pasul la care a eșuat.

PHP

Queuety::retry_workflow( $workflow_id );

Aruncă RuntimeException dacă fluxul de lucru a fost deja compensat după o eroare.

`Queuety::pause_workflow( $workflow_id )`

Pune pe pauză un flux de lucru în curs de execuție. Lucrările deja puse în coadă pentru pasul curent se pot finaliza, dar pașii următori nu sunt puși în coadă până când nu reiei fluxul de lucru.

PHP

Queuety::pause_workflow( $workflow_id );

`Queuety::resume_workflow( $workflow_id )`

Reia un flux de lucru pus pe pauză.

PHP

Queuety::resume_workflow( $workflow_id );

`Queuety::cancel_workflow( $workflow_id )`

Anulează un flux de lucru și execută toate rutinele de curățare înregistrate. Setează starea la cancelled și îngroapă toate lucrările în așteptare și în curs de procesare pentru fluxul de lucru.

PHP

Queuety::cancel_workflow( $workflow_id );

Aruncă RuntimeException dacă fluxul de lucru este deja completed sau cancelled. Consultați Anulare pentru detalii.

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

Derulează un flux de lucru înapoi la un pas anterior și îl execută din nou de acolo. Starea este restaurată din jurnalul de evenimente state_after pentru pasul respectiv, iar pasul următor este pus în coadă.

PHP

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

Aruncă RuntimeException dacă nu există nicio stare înregistrată pentru pasul țintă. Necesită Jurnalul de evenimente al fluxului de lucru. Consultați Călătorie în timp pentru detalii.

`Queuety::fork_workflow( $workflow_id )`

Ramifică un flux de lucru în curs de execuție într-o copie independentă, pornind de la starea curentă. Returnează ID-ul noului flux de lucru.

PHP

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

Fluxul de lucru ramificat primește propriul ID și funcționează independent de original. Consultați Ramificare pentru detalii.

`Queuety::export_workflow( $workflow_id )`

Exportă întregul istoric de execuție al unui flux de lucru ca array serializabil JSON. Include rândul fluxului de lucru, toate lucrările, evenimentele fluxului și intrările din jurnal.

PHP

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

Consultați Export și reluare pentru structura JSON și cazuri de utilizare.

`Queuety::replay_workflow( $data )`

Reia un flux de lucru exportat în mediul curent. Creează un flux de lucru nou din datele de export și returnează ID-ul acestuia.

PHP

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

Consultați Export și reluare pentru detalii.

`Queuety::replace_wp_cron()`

Înlocuiește sistemul încorporat WordPress cron cu programatorul de sarcini al Queuety. Interceptează wp_schedule_event(), wp_unschedule_event() și wp_get_scheduled_event() și direcționează evenimentele recurente prin lucrătorul Queuety.

PHP

Queuety::replace_wp_cron();

Funcționează numai când este încărcat WordPress. Consultați Înlocuirea WP-Cron pentru detalii.

`Queuety::restore_wp_cron()`

Restaurează comportamentul implicit WordPress cron eliminând hook-urile punții Queuety cron.

PHP

Queuety::restore_wp_cron();

Consultați Înlocuirea WP-Cron pentru limitări.

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

Trimite un semnal unui flux de lucru. Dacă fluxul așteaptă în prezent acest semnal, se reia imediat. În caz contrar, semnalul este stocat și preluat când fluxul ajunge la pasul wait_for_signal corespunzător.

PHP

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

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

Consultați Semnale pentru fluxuri de lucru pentru detalii.

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

Înveliș semantic în jurul signal() pentru fluxuri de aprobare.

PHP

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

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

Înveliș semantic în jurul signal() pentru fluxuri de respingere.

PHP

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

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

Înveliș semantic în jurul signal() pentru date structurate introduse de oameni.

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 = [] )`

Stochează un artefact durabil pentru un flux de lucru.

PHP

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

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

Stochează un artefact pentru pasul fluxului de lucru aflat în curs de execuție.

PHP

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

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

Preia un artefact stocat pentru un flux de lucru.

PHP

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

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

Listează artefactele unui flux de lucru.

PHP

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

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

Șterge un artefact dintr-un flux de lucru.

PHP

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

Mașini de stări

`Queuety::machine( $name )`

Începeți construirea unei mașini de stări. Returnează un 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 ] );

Metode `StateMachineBuilder`

Metodă	Descriere
`initial( string $state_name )`	Setează explicit starea inițială.
`state( string $state_name, ?StateMachineStatus $terminal_status = null )`	Începe configurarea unei stări.
`action( string\|array $action_class )`	Înregistrează o acțiune de intrare pusă în coadă pentru starea curentă. Transmite un nume de clasă sau o definiție structurată cu `class`, `payload` și `config`; `payload` ajunge la `StateAction::handle()` ca `$payload`.
`on( string $event, string $target, string\|array\|null $guard_class = null, ?string $name = null )`	Înregistrează o tranziție de eveniment din starea curentă. Condiția de validare acceptă aceeași formă structurată ca `action()`.
`on_queue( string $queue )`	Setează coada folosită pentru acțiunile de stare puse în coadă.
`with_priority( Priority $priority )`	Setează prioritatea acțiunilor de stare puse în coadă.
`max_attempts( int $max_attempts )`	Setează limita de reîncercări pentru acțiunile de stare puse în coadă.
`version( string $version )`	Etichetează definiția mașinii cu o versiune la nivel de aplicație.
`idempotency_key( string $key )`	Face distribuirea durabilă pentru o cheie furnizată de apelant. O nouă distribuire cu aceeași cheie returnează ID-ul inițial al mașinii.
`dispatch( array $initial_state = [] )`	Distribuie mașina și returnează ID-ul acesteia.

Clasele de acțiuni la intrarea într-o stare pot expune și o metodă opțională config() cu aceleași chei pentru politica de resurse pe care le acceptă pașii fluxului de lucru: concurrency_group, concurrency_limit și cost_units.

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

Distribuie un pachet serializat cu definiția unei mașini de stări. Majoritatea codului aplicației ar trebui să prefere machine()->dispatch(). Această metodă de nivel inferior există pentru ca adaptoarele să poată porni o mașină fără să depindă direct de constructor.

PHP

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

`Queuety::machine_status( $machine_id )`

Obține starea curentă a mașinii. Returnează un obiect StateMachineState sau 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 expune și:

definition_version
definition_hash
idempotency_key
error_message
current_action
terminal_status

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

Trimite un eveniment extern către o mașină aflată în așteptare.

PHP

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

Datele utile sunt disponibile condițiilor de validare a tranzițiilor și sunt îmbinate în starea publică a mașinii când evenimentul este acceptat.

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

Listează rândurile persistate ale mașinii pentru inspecție.

PHP

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

Fiecare rând include ID-ul mașinii, numele, starea, starea curentă, versiunea definiției, hash-ul definiției, cheia de idempotență și marcajele temporale ale ciclului de viață.

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

Obține cronologia durabilă a evenimentelor pentru o mașină.

PHP

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

Fiecare rând de eveniment conține: 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 și created_at.

`Queuety::machine_trace( $machine_id )`

Obține un pachet de urmărire normalizat pentru un UI de depanare a mașinilor de stări.

PHP

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

Urmărirea conține machine, grupate states, brute events, asociate acțiunii jobs, logs, artifacts și chunks.

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

Obține starea publică după un anumit eveniment din urmărirea mașinii sau null dacă nu este găsită.

PHP

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

Consultați Mașini de stări pentru modelul ciclului de viață și tipare de utilizare.

Șabloane de flux de lucru

`Queuety::define_workflow( $name )`

Creează un constructor de fluxuri de lucru care poate fi folosit ca șablon.

PHP

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

`Queuety::register_workflow_template( $builder )`

Înregistrează un șablon de flux de lucru.

PHP

Queuety::register_workflow_template( $builder );

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

Distribuie după nume un șablon de flux de lucru înregistrat. Returnează ID-ul fluxului de lucru.

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' ],
);

Aruncă RuntimeException dacă șablonul nu este înregistrat.

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

Distribuie un flux de lucru dintr-un pachet cu definiția de execuție.

Majoritatea codului aplicației ar trebui să prefere run_workflow() sau workflow()->dispatch(). Această metodă de nivel inferior permite adaptoarelor reutilizabile să distribuie o definiție serializată a fluxului de lucru fără să depindă de metodele interne ale managerului de fluxuri de lucru.

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

Înregistrează o acțiune WordPress care distribuie un flux de lucru.

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 poate fi fie:

numele unui șablon de flux de lucru înregistrat
un WorkflowBuilder în linie

map, when și idempotency_key rulează sincron în cadrul cererii WordPress curente. Numai pașii fluxului de lucru rulează asincron după distribuire.

Consultați Declanșatoare de acțiuni pentru ghidul complet.

`Queuety::workflow_templates()`

Obține registrul șabloanelor de flux de lucru.

PHP

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

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

Încarcă și înregistrează toate fișierele de flux de lucru dintr-un director. Fiecare fișier .php trebuie să returneze un WorkflowBuilder. Returnează numărul fluxurilor de lucru înregistrate.

PHP

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

`Queuety::load_workflow_file( $file_path )`

Încarcă și înregistrează un singur fișier de flux de lucru. Returnează WorkflowTemplate sau null dacă fișierul nu returnează un constructor.

PHP

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

Programare

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

Creează un program recurent. Returnează un constructor PendingSchedule.

PHP

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

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

Metode `PendingSchedule`

Metodă	Descriere
`every( string $interval )`	Setează o expresie de interval (e.g. `1 hour`, `30 minutes`)
`cron( string $expression )`	Setează o expresie cron cu 5 câmpuri
`overlap( OverlapPolicy $policy )`	Setează politica de suprapunere (implicit: `Allow`). Consultați Programare.
`on_queue( string $queue )`	Setează coada țintă (implicit: `default`)
`id()`	Forțează distribuirea și returnează ID-ul programării

`Queuety::scheduler()`

Obține instanța internă Scheduler.

PHP

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

Înregistrarea gestionarilor

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

Înregistrează o clasă de gestionar sub un nume.

PHP

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

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

Descoperă automat și înregistrează clasele de gestionari dintr-un director. Returnează numărul gestionarilor înregistrați.

PHP

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

`Queuety::registry()`

Obține registrul gestionarilor.

PHP

$registry = Queuety::registry();

Observabilitate

`Queuety::logger()`

Obține instanța Logger pentru interogarea intrărilor din jurnal.

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

Obține instanța Metrics.

PHP

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

`Queuety::webhook_notifier()`

Obține instanța WebhookNotifier.

PHP

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

`Queuety::rate_limiter()`

Obține instanța RateLimiter.

PHP

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

Cache

`Queuety::cache()`

Obține instanța activă a backendului de cache.

PHP

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

`Queuety::set_cache( $cache )`

Suprascrie backendul de cache. Apelează înainte de Queuety::init() pentru a folosi o implementare personalizată.

PHP

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

`QUEUETY_CACHE_TTL`

TTL implicit al cache-ului, în secunde (implicit: 5). Setează această constantă în wp-config.php pentru a schimba TTL-ul implicit folosit de operațiile interne ale cache-ului.

PHP

define( 'QUEUETY_CACHE_TTL', 10 );

Consultați Memorarea în cache pentru detalii despre backends și detectarea automată.

Jurnalul de evenimente al mașinii de stări

`Queuety::machine_events()`

Obține instanța StateMachineEventLog.

PHP

$log = Queuety::machine_events();

Jurnalul de evenimente al fluxului de lucru

`Queuety::workflow_events()`

Obține instanța WorkflowEventLog.

PHP

$log = Queuety::workflow_events();

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

Obține cronologia completă a evenimentelor unui flux de lucru. Returnează un array de rânduri de evenimente, ordonate după ID.

PHP

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

Fiecare rând de eveniment conține: 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 (copiat din error.message), duration_ms și created_at.

`Queuety::workflow_trace( $workflow_id )`

Obține un pachet de urmărire normalizat pentru un UI de depanare a fluxurilor de lucru.

PHP

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

Urmărirea conține workflow, grupate steps, brute events, jobs, logs, artifacts, chunks, signals și wait_dependencies.

`Queuety::trace_input( $input )`

Setează datele de intrare la nivel de aplicație pentru pasul fluxului de lucru aflat în curs de execuție.

`Queuety::trace_output( $output )`

Setează datele de ieșire la nivel de aplicație pentru pasul fluxului de lucru aflat în curs de execuție.

`Queuety::trace_context( array $context )`

Îmbină contextul la nivel de aplicație în pasul fluxului de lucru aflat în curs de execuție.

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

Obține starea publică după finalizarea sau reluarea unui anumit pas al fluxului de lucru sau null dacă nu este găsită.

PHP

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

Consultați Jurnalul de evenimente al fluxului de lucru pentru cazuri de utilizare și comenzi CLI.

Instanțe interne

Aceste metode returnează instanțe interne pentru cazuri de utilizare avansate.

Metodă	Returnează	Descriere
`Queuety::queue()`	`Queue`	Operații asupra cozii (revendicare, eliberare, îngropare)
`Queuety::worker()`	`Worker`	Controlul procesului lucrătorului
`Queuety::workflow_manager()`	`Workflow`	Orchestrarea fluxurilor de lucru
`Queuety::machines()`	`StateMachine`	Gestionarea ciclului de viață al mașinii de stări
`Queuety::connection()`	`Connection`	Conexiunea la baza de date
`Queuety::batch_manager()`	`BatchManager`	Gestionarea ciclului de viață al loturilor (anulare, curățare)
`Queuety::chunk_store()`	`ChunkStore`	Persistarea fragmentelor pentru pașii de streaming
`Queuety::machine_events()`	`StateMachineEventLog`	Evenimente de urmărire a mașinii de stări și pachete de urmărire normalizate
`Queuety::workflow_events()`	`WorkflowEventLog`	Evenimente de urmărire a fluxului de lucru și pachete de urmărire normalizate
`Queuety::resource_manager()`	`ResourceManager`	Profiluri de admitere a lucrătorilor, bugete ponderate și inspecția memoriei sistemului
`Queuety::cache()`	`Cache`	Instanța backendului de cache

Testare

`Queuety::fake()`

Înlocuiește coada cu un substitut în memorie pentru testare. Returnează o instanță QueueFake care înregistrează lucrările distribuite fără a accesa baza de date. create_batch() înregistrează și metadatele loturilor fictive; lanțurile sunt verificate prin lucrările individuale distribuite.

PHP

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

`Queuety::reset()`

Resetează toată starea internă, inclusiv substitutul cozii. Apelează în tearDown().

PHP

Queuety::reset();

Metode `QueueFake`

Metodă	Descriere
`assert_pushed( string $class, ?Closure $callback )`	Verifică dacă a fost pusă în coadă o clasă de lucrări. Un callback opțional filtrează după datele lucrării.
`assert_pushed_times( string $class, int $count )`	Verifică dacă o lucrare a fost pusă în coadă de exact N ori.
`assert_not_pushed( string $class )`	Verifică dacă o lucrare nu a fost pusă niciodată în coadă.
`assert_nothing_pushed()`	Verifică dacă nu a fost pusă nicio lucrare în coadă.
`assert_batched( ?Closure $callback )`	Verifică dacă a fost distribuit un lot. Un callback opțional filtrează după datele lotului.
`pushed( string $class )`	Obține toate distribuirile înregistrate pentru o clasă. Returnează un array.
`batches()`	Obține toate loturile înregistrate. Returnează un array.
`reset()`	Șterge toate lucrările și loturile înregistrate.

Consultați Testare pentru exemple complete.

Interfețe

`Handler`

Interfață pentru gestionari simpli de lucrări.

PHP

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

config() poate returna queue, max_attempts, backoff, rate_limit, concurrency_group, concurrency_limit și cost_units.

`Step`

Interfață pentru gestionarii pașilor din fluxurile de lucru.

PHP

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

config() poate returna max_attempts, backoff, concurrency_group, concurrency_limit și cost_units.

`Contracts\ForEachHandler`

Interfață pentru gestionarii ramurilor dinamice de tip for-each.

PHP

namespace Queuety\Contracts;

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

config() poate returna max_attempts, backoff, rate_limit, concurrency_group, concurrency_limit și cost_units.

`Contracts\ForEachReducer`

Reductor opțional pentru rezultatele for-each.

PHP

namespace Queuety\Contracts;

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

`Contracts\Compensation`

Interfață pentru gestionarii de compensare a pașilor.

PHP

namespace Queuety\Contracts;

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

`Contracts\RepeatCondition`

Interfață pentru predicatele buclelor folosite de repeat_until() și repeat_while() când o comparație între cheile de stare este prea limitată.

PHP

namespace Queuety\Contracts;

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

`Contracts\StateAction`

Interfață pentru acțiunile de intrare în stări puse în coadă.

PHP

namespace Queuety\Contracts;

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

$payload primește valoarea structurată payload din definiția acțiunii, dacă mașina a fost definită cu aceasta. Returnează actualizări ale stării publice. Includeți _event și, opțional, _event_payload când acțiunea trebuie să declanșeze imediat o tranziție.

Acțiunile de stare pot defini și o metodă opțională config(). Queuety preia din aceasta concurrency_group, concurrency_limit și cost_units când acțiunea de intrare este pusă în coadă.

`Contracts\StateGuard`

Interfață pentru condițiile de validare a tranzițiilor mașinii de stări.

PHP

namespace Queuety\Contracts;

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

`Contracts\Job`

Interfață pentru clase de lucrări autonome care pot fi distribuite. Clasele de lucrări își încapsulează datele utile în proprietăți publice și logica de execuție în handle().

PHP

namespace Queuety\Contracts;

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

Clasele de lucrări pot defini opțional o metodă middleware() care returnează un array de instanțe Contracts\Middleware.

`Contracts\Cache`

Interfață pentru backends de cache. Consultați Memorarea în cache pentru implementările încorporate și backends personalizate.

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;
}

Metodă	Returnează	Descriere
`get( $key )`	`mixed`	Preia o valoare. Returnează `null` dacă nu este găsită sau a expirat.
`set( $key, $value, $ttl )`	`void`	Stochează o valoare. Un TTL de 0 înseamnă că valoarea nu expiră.
`delete( $key )`	`void`	Elimină o valoare.
`has( $key )`	`bool`	Verifică dacă o cheie există și nu a expirat.
`add( $key, $value, $ttl )`	`bool`	Setează atomic o valoare numai dacă nu există. Returnează `true` dacă valoarea a fost setată.
`flush()`	`void`	Elimină toate elementele din cache.

`WorkflowEventLog`

Înregistrează tranzițiile pașilor din fluxul de lucru cu contextul complet al urmăririi. Accesează prin Queuety::workflow_events().

Metodă	Returnează	Descriere
`record_event( array $event )`	`void`	Înregistrează un eveniment brut de urmărire
`record_step_started( ... )`	`void`	Înregistrează un eveniment step_started cu datele de intrare și starea dinaintea execuției
`record_step_completed( ... )`	`void`	Înregistrează un eveniment step_completed cu datele de intrare, rezultatul, starea dinainte și cea de după
`record_step_failed( ... )`	`void`	Înregistrează un eveniment step_failed cu contextul urmăririi și detalii structurate despre eroare
`get_timeline( int $workflow_id, ?int $limit = 100, int $offset = 0 )`	`array`	Obține evenimentele de urmărire ale fluxului de lucru în ordine, cu paginare opțională
`get_trace( int $workflow_id )`	`array`	Obține un pachet de urmărire normalizat
`get_state_at_step( int $workflow_id, int $step_index )`	`array\|null`	Obține starea publică după finalizarea sau reluarea unui anumit pas
`prune( int $older_than_days )`	`int`	Șterge evenimentele mai vechi decât numărul de zile specificat. Returnează numărul de rânduri șterse.

`StateMachineEventLog`

Înregistrează evenimentele ciclului de viață al mașinii de stări, acțiunile, tranzițiile și condițiile de validare, cu aceleași câmpuri de urmărire ca evenimentele fluxurilor de lucru.

Metodă	Returnează	Descriere
`record_event( array $event )`	`void`	Înregistrează un eveniment brut de urmărire a mașinii
`get_timeline( int $machine_id, ?int $limit = 100, int $offset = 0 )`	`array`	Obține evenimentele de urmărire a mașinii în ordine, cu paginare opțională
`get_trace( int $machine_id )`	`array`	Obține un pachet normalizat de urmărire a mașinii
`get_state_at_event( int $machine_id, int $event_id )`	`array\|null`	Obține starea publică după un anumit eveniment de urmărire a mașinii
`prune( int $older_than_days )`	`int`	Șterge evenimentele mai vechi decât numărul de zile indicat. Returnează numărul de rânduri șterse.

`Contracts\StreamingStep`

Interfață pentru gestionarele pașilor de flux de lucru care produc un flux de fragmente. Fiecare valoare emisă este stocată imediat în baza de date. Consultați Pași de flux.

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;
}

Metodă	Descriere
`stream( $state, $existing_chunks )`	Generator care emite fragmente de șiruri. Fiecare emisie declanșează imediat o scriere în baza de date. La reîncercare, `$existing_chunks` conține fragmentele stocate anterior.
`on_complete( $chunks, $state )`	Apelată după încheierea fluxului. Returnează datele care vor fi îmbinate în starea fluxului de lucru.
`config()`	Configurare opțională. Acceptă `max_attempts`, `backoff`, `concurrency_group`, `concurrency_limit` și `cost_units`.

`ChunkStore`

Gestionează stocarea fragmentelor pentru pașii de flux. Accesați prin Queuety::chunk_store().

Metodă	Returnează	Descriere
`get_chunks( int $job_id )`	`string[]`	Preia toate fragmentele unei sarcini, ordonate după indexul fragmentului
`append_chunk( int $job_id, int $chunk_index, string $content, ?int $workflow_id, ?int $step_index )`	`void`	Adaugă un singur fragment
`clear_chunks( int $job_id )`	`void`	Șterge toate fragmentele unei sarcini
`chunk_count( int $job_id )`	`int`	Numără fragmentele unei sarcini
`get_accumulated( int $job_id )`	`string`	Concatenează toate fragmentele într-un singur șir

`Heartbeat`

Metodă statică ajutătoare pentru trimiterea bătăilor de activitate din interiorul gestionarilor de pași și al pașilor de flux. Consultați Bătăi de activitate.

Metodă	Descriere
`Heartbeat::init( int $job_id, Connection $conn )`	Setează contextul sarcinii. Apelată automat de lucrător.
`Heartbeat::beat( array $progress = [] )`	Trimite o bătaie de activitate. Actualizează `reserved_at` și, opțional, stochează datele de progres în `heartbeat_data`.
`Heartbeat::clear()`	Șterge contextul sarcinii. Apelată automat de lucrător.
`Heartbeat::current_job_id()`	Obține ID-ul sarcinii curente (pentru testare).

`Contracts\Middleware`

Interfață pentru middleware-ul sarcinilor. Middleware-ul încadrează execuția sarcinii într-o conductă în stilul unei cepe.

PHP

namespace Queuety\Contracts;

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

Trăsături

`Dispatchable`

Oferă metode statice de expediere pentru clasele de sarcini care implementează Contracts\Job. Argumentele constructorului sunt transmise pentru a crea instanța sarcinii.

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' );

Metodă	Returnează	Descriere
`dispatch( ...$args )`	`PendingJob`	Creează o instanță și o expediază
`dispatch_if( bool $condition, ...$args )`	`PendingJob\|null`	Expediază doar dacă `$condition` este true
`dispatch_unless( bool $condition, ...$args )`	`PendingJob\|null`	Expediază doar dacă `$condition` este false
`with_chain( array $jobs )`	`ChainBuilder`	Pornește un lanț din această clasă de sarcină

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

Proprietățile sarcinii

Clasele de sarcini pot declara proprietăți publice pe care lucrătorul le citește prin reflecție pentru a suprascrie comportamentul de reîncercare și de expirare a timpului.

Proprietate	Tip	Descriere
`$tries`	`int`	Numărul maxim de încercări înainte ca sarcina să fie îngropată
`$timeout`	`int`	Timpul maxim de execuție în secunde (necesită `pcntl`)
`$backoff`	`array`	Valori de întârziere progresivă, în secunde, pentru fiecare încercare de reîncercare

PHP

class ImportProductsJob implements Job {
    use Dispatchable;

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

    // ...
}

Cârligul `failed()`

Când o sarcină este îngropată după epuizarea tuturor reîncercărilor, lucrătorul apelează o metodă failed() pe instanța sarcinii, dacă este definită.

PHP

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

Middleware încorporat

Toate clasele middleware implementează Contracts\Middleware și se află în spațiul de nume Queuety\Middleware.

Clasă	Descriere
`RateLimited( int $max, int $window )`	Impune o limită de rată de `$max` execuții la fiecare `$window` secunde
`Timeout( int $seconds )`	Oprește sarcina după `$seconds` secunde (necesită `pcntl`)
`UniqueJob( ?string $key )`	Împiedică execuția simultană printr-o blocare în baza de date
`WithoutOverlapping( string $key, int $release_after )`	Împiedică suprapunerea execuțiilor, cu expirarea automată a blocării
`ThrottlesExceptions( int $max_attempts, int $decay_minutes )`	Limitează excepțiile pentru a preveni valurile de sarcini atunci când un serviciu extern nu funcționează

Consultați Middleware pentru detalii și exemple de utilizare.

Enumerații

`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';
}

Consultați Programare pentru detalii despre funcționarea fiecărei politici.

Atribute

`#[QueuetyHandler]`

Atributul PHP 8 pentru înregistrare automată. Consultați PHP Atribute.

PHP

use Queuety\Attributes\QueuetyHandler;

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

Puntea CLI

`Queuety::cli_command()`

Returnează numele publicat al comenzii rădăcină WP-CLI.

Comanda rădăcină implicită este queuety. Definiți QUEUETY_CLI_COMMAND înainte de pornirea lui Queuety dacă doriți ca această comandă să fie publicată sub un alt nume.

`Queuety::cli_command_map()`

Returnează catalogul serializabil Queuety CLI pentru mediile de testare care trebuie să cunoască interfața comenzilor livrate fără a analiza clasele WP-CLI.

Fiecare intrare include:

un ID de operație stabil
calea relativă CLI
clasa și metoda gestionarului WP-CLI
țintele posibile de execuție
adaptorul care normalizează argumentele brute CLI într-un plan de execuție

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

Transformă o invocare CLI analizată într-un plan de execuție pe care mediul de testare ar trebui să îl ruleze.

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'] );

Planurile de execuție țin cont în mod intenționat de transport:

comenzile se rezolvă în funcții apelabile publice PHP, cu argumente normalizate
catalogul enumeră în avans toate țintele de execuție posibile, în timp ce callable și transport identifică ramura aleasă pentru intrarea analizată

Acesta este modelul standard Queuety folosit de mediile de testare pentru agenți: ID de operație stabil, țintă publică API și normalizare gestionată de adaptor pentru aspecte specifice doar CLI, precum valorile implicite ale cozii, decodarea payloadului JSON, ramificarea grupului de lucrători și gestionarea fișierelor de export ale fluxului de lucru.

PHP API

Acces autonom la nucleu

Queuety::init( Connection $conn )

Queuety::ensure_schema()

Sarcini

Queuety::dispatch( $handler, $payload )

Metode PendingJob

Queuety::dispatch_sync( $handler, $payload )

Queuety::batch( $jobs )

Queuety::create_batch( $jobs )

Metode BatchBuilder

Queuety::find_batch( $id )

Controlul workerilor

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

Obiectul valoare Batch

Queuety::chain( $jobs )

Metode ChainBuilder

Queuety::retry( $job_id )

Queuety::retry_buried()

Queuety::purge( $older_than_days )

Cozi

Queuety::stats( $queue )

Queuety::buried( $queue )

Queuety::pause( $queue )

Queuety::resume( $queue )

Queuety::is_paused( $queue )

Fluxuri de lucru

Queuety::workflow( $name )

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