PHP API

Peržiūrėti kaip Markdown Redaguoti per GitHub Dokumentacija: 2cc7b2b

Visa Queuety funkcionalumas pasiekiamas per statinę klasę Queuety\Queuety. Tai vienintelis įėjimo taškas užduotims išsiųsti, darbo eigoms kurti, būsenų automatams vykdyti, eilėms tvarkyti ir prieigai prie vidinių posistemių.

PHP

use Queuety\Queuety;

Atskira prieiga prie pagrindinių komponentų

Queuety DB pagrįstą vykdymo aplinką galima inicijuoti už WordPress ribų.

Šis atskiro naudojimo būdas skirtas orkestravimo pagrindui:

užduotims ir vykdytojams (vykdytojų telkiniams taip pat reikia DB_HOST, DB_NAME, DB_USER ir DB_PASSWORD konstantų; žr. toliau)
darbo eigoms ir signalams
būsenų automatams ir įvykiams
artefaktams, žurnalams, metrikoms ir tvarkaraščiams

Tai nepakeičia WordPress adapterio. Queuety::on_action(), replace_wp_cron(), papildinio įkrovos procesui ir tikrajam wp queuety ... registravimo keliui vis dar reikia WordPress.

`Queuety::init( Connection $conn )`

Inicijuokite Queuety tiesioginiu ryšiu su 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 );

Pasirinktinis argumentas table_prefix keičia vykdymo aplinkos lentelių pavadinimų Queuety dalį kiekvienam ryšiui:

PHP

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

Taip tokios lentelės kaip queuety_jobs susiejamos su wp_themequeue_jobs.

`Queuety::ensure_schema()`

Įsitikinkite, kad aktyviam ryšiui skirtos Queuety vykdymo aplinkos lentelės yra sukurtos.

PHP

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

Visą modelį ir WordPress ribas žr. dokumente Atskiras naudojimas.

Užduotys

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

Išsiųskite užduotį. Priimamas tvarkyklės pavadinimas kaip eilutė arba Contracts\Job egzempliorius. Grąžinamas PendingJob kūrimo objektas, leidžiantis grandininiu būdu nustatyti parinktis.

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

Jei perduodamas Contracts\Job egzempliorius, serializatorius ištraukia FQCN kaip tvarkyklės pavadinimą, o viešąsias savybes – kaip naudingąjį turinį. Parametras $payload ignoruojamas, kai $handler yra Job egzempliorius.

`PendingJob` metodai

Metodas	Aprašymas
`on_queue( string $queue )`	Nustatyti paskirties eilę
`with_priority( Priority $priority )`	Nustatyti prioritetą
`delay( int $seconds )`	Atidėjimas iki užduotis taps prieinama
`max_attempts( int $max )`	Didžiausias bandymų skaičius
`rate_limit( int $max, int $window )`	Dažnio ribojimas: didžiausias vykdymų skaičius per lango sekundžių skaičių
`unique()`	Neleisti išsiųsti pasikartojančių užduočių
`after( int $job_id )`	Palaukti, kol pirma bus baigta kita užduotis
`concurrency_group( string $group, int $limit )`	Įtraukti užduotį į bendrą vykdytojų priėmimo grupę
`cost_units( int $units )`	Prie išsiuntimo pridėti santykinio vykdymo sąnaudų metaduomenis
`id()`	Priverstinai išsiųsti užduotį ir grąžinti jos ID

concurrency_group() ir cost_units() naudojami platesniame Išteklių valdymas modelyje, todėl tas pats išsiuntimas gali būti įtrauktas į pagal skaičių taikomus lygiagretaus vykdymo limitus, svertinius eilės ar teikėjo biudžetus ir vykdytojų priėmimo euristikas.

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

Vykdyti užduotį sinchroniškai dabartiniame procese, neįtraukiant jos į eilę.

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

Užduoties metodas handle() iškviečiamas tiesiogiai. Duomenų bazės eilutė nesukuriama, vykdytojas nedalyvauja, o tarpinė programinė įranga netaikoma.

`Queuety::batch( $jobs )`

Išsiųsti kelias užduotis vienu kelių eilučių INSERT sakiniu.

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

Sukurkite BatchBuilder, kad išsiųstumėte užduočių grupę su iškvietimo funkcijomis ir eigos stebėjimu. Išsamią informaciją žr. skyriuje Grupinis vykdymas.

PHP

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

`BatchBuilder` metodai

Metodas	Grąžinama reikšmė	Aprašymas
`name( string $name )`	`self`	Nustatyti grupės pavadinimą
`on_queue( string $queue )`	`self`	Nustatyti visų grupės užduočių eilę
`then( string $handler_class )`	`self`	Tvarkyklė, iškviečiama sėkmingai užbaigus grupę
`catch( string $handler_class )`	`self`	Tvarkyklė, iškviečiama nepavykus pirmajai grupės užduočiai
`finally( string $handler_class )`	`self`	Tvarkyklė, iškviečiama grupei pasibaigus (sėkmingai arba nesėkmingai)
`allow_failures()`	`self`	Leisti iškviesti funkciją `then` net ir įvykus klaidoms
`dispatch()`	`Batch`	Sukurti grupės eilutę ir visas užduotis; grąžinti `Batch` reikšmės objektą

`Queuety::find_batch( $id )`

Rasti grupę pagal ID. Grąžinamas Batch reikšmės objektas arba null.

PHP

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

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

Vykdytojų valdymas

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

Tiesiogiai vykdyti vieną vykdytojo ciklą.

PHP

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

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

Tiesiogiai paleisti fiksuoto dydžio vykdytojų telkinį. Reikalingi plėtiniai pcntl ir posix. Telkinys antriniams ryšiams nuskaito konstantas WordPress DB_HOST, DB_NAME, DB_USER ir DB_PASSWORD (iš inicijuoto ryšio perimami tik lentelių prefiksai), todėl paleisdami telkinį už WordPress ribų jas apibrėžkite (#5).

PHP

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

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

Tiesiogiai paleisti adaptyvų vykdytojų telkinį. Reikalavimai tokie pat kaip run_worker_pool().

PHP

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

`Batch` reikšmės objektas

Metodas / savybė	Tipas	Aprašymas
`$id`	`int`	Grupės ID
`$name`	`string\|null`	Pasirinktinis grupės pavadinimas
`$total_jobs`	`int`	Bendras užduočių skaičius grupėje
`$pending_jobs`	`int`	Dar laukiančios užduotys
`$failed_jobs`	`int`	Nepavykusių užduočių skaičius
`$failed_job_ids`	`array`	Nepavykusių užduočių ID
`$created_at`	`DateTimeImmutable`	Grupės sukūrimo laikas
`$finished_at`	`DateTimeImmutable\|null`	Grupės pabaigos laikas
`$cancelled_at`	`DateTimeImmutable\|null`	Grupės atšaukimo laikas
`progress()`	`int`	Užbaigimo procentinė dalis (0-100)
`finished()`	`bool`	Ar grupė užbaigta
`has_failures()`	`bool`	Ar nepavyko kuri nors užduotis
`cancelled()`	`bool`	Ar grupė atšaukta

`Queuety::chain( $jobs )`

Sukurkite ChainBuilder nuosekliam užduočių vykdymui. Kiekviena užduotis priklauso nuo to, ar ankstesnė užduotis sėkmingai užbaigta.

PHP

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

`ChainBuilder` metodai

Metodas	Grąžinama reikšmė	Aprašymas
`on_queue( string $queue )`	`self`	Nustatyti visų grandinės užduočių eilę
`catch( string $handler_class )`	`self`	Tvarkyklė, iškviečiama, kai kuri nors grandinės užduotis visam laikui perkeliama į saugyklą
`dispatch()`	`int`	Sukurti visas užduotis su nuosekliomis priklausomybėmis; grąžinti pirmosios užduoties ID

`Queuety::retry( $job_id )`

Pakartotinai vykdyti konkrečią užduotį. Atkuriama jos pending būsena.

PHP

Queuety::retry( 42 );

`Queuety::retry_buried()`

Pakartotinai vykdyti visas saugykloje esančias užduotis. Grąžinamas pakartotinai vykdytų užduočių skaičius.

PHP

$count = Queuety::retry_buried();

`Queuety::purge( $older_than_days )`

Išvalyti užbaigtas užduotis. Grąžinamas išvalytų užduočių skaičius.

PHP

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

Eilės

`Queuety::stats( $queue )`

Gauti užduočių skaičių pagal būseną.

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

Gauti visas saugykloje esančias užduotis.

PHP

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

`Queuety::pause( $queue )`

Pristabdyti eilę, kad vykdytojai ją praleistų.

PHP

Queuety::pause( 'emails' );

`Queuety::resume( $queue )`

Tęsti pristabdytą eilę.

PHP

Queuety::resume( 'emails' );

`Queuety::is_paused( $queue )`

Patikrinti, ar eilė pristabdyta.

PHP

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

Darbo eigos

`Queuety::workflow( $name )`

Pradėti kurti darbo eigą. Grąžinamas WorkflowBuilder.

PHP

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

`WorkflowBuilder` metodai

Metodas	Aprašymas
`then( string $handler_class, ?string $name = null )`	Pridėti nuoseklų veiksmą. Sąlyginio šakojimo pavadinimas pasirinktinis.
`parallel( array $handler_classes )`	Pridėti lygiagrečių veiksmų grupę.
`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 )`	Pridėti dinaminį ciklo per kiekvieną elementą veiksmą, kuris darbo eigos būsenoje kiekvienam elementui sukuria atskirą šakos užduotį.
`repeat_while( string $target_step, ?string $state_key = null, mixed $expected = true, ?string $name = null, ?string $condition_class = null, ?int $max_iterations = null )`	Grįžti prie ankstesnio pavadinto veiksmo, kol atitinka viešasis būsenos raktas arba klasė `RepeatCondition`.
`repeat_until( string $target_step, ?string $state_key = null, mixed $expected = true, ?string $name = null, ?string $condition_class = null, ?int $max_iterations = null )`	Grįžti prie ankstesnio pavadinto veiksmo, kol atitiks viešasis būsenos raktas arba klasė `RepeatCondition`.
`run_workflow( string $workflow_name, WorkflowBuilder $sub_builder )`	Pridėti darbo eigos vykdymo veiksmą.
`delay( int $seconds = 0, int $minutes = 0, int $hours = 0, int $days = 0 )`	Pridėti išliekantį delsos veiksmą. Sumavamos visos trukmės.
`wait_for_signal( string $name, ?string $result_key = null, ?string $step_name = null, array $match_payload = [], ?string $correlation_key = null )`	Pridėti laukimo signalo veiksmą. Pasirinktiniai `match_payload` ir `correlation_key` leidžia nepaisyti nesusijusių įvykių.
`wait_for_signals( array $signal_names, WaitMode $mode = WaitMode::All, ?string $result_key = null, ?string $name = null, array $match_payload = [], ?string $correlation_key = null )`	Laukti kelių signalų: visų arba bet kurio iš jų; galima nurodyti naudingąjį turinį ir koreliaciją.
`wait_for_approval( string $signal_name = 'approval', ?string $result_key = 'approval', ?string $name = null, array $match_payload = [], ?string $correlation_key = null )`	Laukti žmogaus patvirtinimo signalo.
`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 )`	Laukti patvirtinimo arba atmetimo ir išsaugoti struktūrizuotą sprendimo naudingąjį turinį.
`wait_for_input( string $signal_name = 'input', ?string $result_key = 'input', ?string $name = null, array $match_payload = [], ?string $correlation_key = null )`	Laukti žmogaus įvesties signalo.
`wait_for_workflow( int\|string $workflow, ?string $result_key = null, ?string $name = null )`	Laukti, kol užbaigs darbą viena darbo eigos ID arba viešojoje būsenoje išsaugota darbo eigos ID.
`wait_for_workflows( array\|string $workflows, WaitMode $mode = WaitMode::All, ?string $result_key = null, ?string $name = null, ?int $quorum = null )`	Laukti, kol užbaigs darbą visos, bet kuri arba kvorumas iš kelių darbo eigų.
`wait_for_workflow_group( string $group_key, WaitMode $mode = WaitMode::All, ?int $quorum = null, ?string $result_key = null, ?string $name = null )`	Laukti anksčiau pradėtos darbo eigų grupės pagal vidinį grupės raktą.
`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 )`	Išsiųskite po vieną aukščiausio lygio darbo eigą kiekvienam vykdymo aplinkos elementui; pradėtą jų grupę pasirinktinai pažymėkite patvariu grupės raktu.
`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 )`	Į agentus orientuotas `start_workflows()` pseudonimas.
`wait_for_agents( array\|string $workflows = 'agent_workflow_ids', WaitMode $mode = WaitMode::All, ?string $result_key = 'agent_results', ?string $name = null, ?int $quorum = null )`	Į agentus orientuotas `wait_for_workflows()` pseudonimas.
`wait_for_agent_group( string $group_key = 'agents', WaitMode $mode = WaitMode::All, ?int $quorum = null, ?string $result_key = 'agent_results', ?string $name = null )`	Į agentus orientuotas `wait_for_workflow_group()` pseudonimas.
`version( string $version )`	Pažymėkite darbo eigos apibrėžtį programos lygmens versija, kad ją būtų galima tikrinti ir eksportuoti.
`idempotency_key( string $key )`	Užtikrinkite patvarų darbo eigos išsiuntimą pagal skambintojo pateiktą raktą. Pakartotinai išsiuntus tą patį raktą, grąžinamas pradinis darbo eigos ID.
`max_transitions( int $max )`	Nepavykdykite darbo eigos, jei užbaigiama daugiau nei `$max` žingsnių perėjimų.
`max_for_each_items( int $max )`	Nepavykdykite darbo eigos, jei `for_each()` žingsnis bando išplėsti daugiau nei `$max` elementų.
`max_state_bytes( int $bytes )`	Nepavykdykite darbo eigos, jei viešoji būsena viršija `$bytes`.
`max_cost_units( int $max )`	Nepavykdykite darbo eigos, jei užbaigti žingsniai sunaudoja daugiau nei `$max` sąnaudų vienetų.
`max_started_workflows( int $max )`	Nepavykdykite darbo eigos prieš išsiųsdami daugiau nei `$max` antrinių darbo eigų.
`compensate_with( string $handler_class )`	Pridėkite kompensavimo apdorojimo funkciją prie paskutinio pridėto žingsnio.
`compensate_on_failure()`	Darbo eigai nepavykus, automatiškai vykdykite užbaigtų žingsnių kompensavimo veiksmus.
`on_queue( string $queue )`	Nustatykite visų žingsnių eilę.
`with_priority( Priority $priority )`	Nustatykite visų žingsnių prioritetą.
`max_attempts( int $max )`	Nustatykite didžiausią pakartotinių bandymų skaičių vienam žingsniui.
`on_cancel( string $handler_class )`	Atšaukimui užregistruokite išvalymo apdorojimo klasę. Išsamiau žr. Atšaukimas.
`prune_state_after( int $steps = 2 )`	Įjunkite automatinį būsenos valymą. Pašalinami senesni už nustatytą ribą žingsnių rezultatai. Išsamiau žr. Būsenos valymas.
`must_complete_within( int $seconds = 0, int $minutes = 0, int $hours = 0, int $days = 0 )`	Nustatykite darbo eigos terminą. Sudedamos visos trukmės. Išsamiau žr. Terminai.
`on_deadline( string $handler_class )`	Užregistruokite apdorojimo funkciją, kuri iškviečiama viršijus darbo eigos terminą. Išsamiau žr. Terminai.
`dispatch( array $initial_payload = [] )`	Išsiųskite darbo eigą. Grąžinamas darbo eigos ID.

`Queuety::workflow_status( $workflow_id )`

Gaukite darbo eigos būseną. Grąžinamas objektas WorkflowState arba 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 taip pat pateikia:

waiting_for: signalų pavadinimus arba darbo eigos ID, kurie šiuo metu stabdo vykdymą
current_step_name: dabartinio žingsnio sukonfigūruotą pavadinimą, jei toks yra
wait_mode: dabartinį laukimo režimą (all, any arba quorum), kol vykdymas sustabdytas
wait_details: papildomus tikrinimo duomenis, pvz., atitikusių ir likusių blokatorių skaičių
definition_version: reikšmę, nustatytą per version(), jei tokia yra
definition_hash: deterministinę darbo eigos apibrėžties maišą
idempotency_key: patvarų išsiuntimo raktą, jei toks yra
budget: sukonfigūruotas ribas ir dabartinį naudojimą, jei įjungtos apsaugos priemonės
artifact_count: darbo eigai saugomų artefaktų skaičių
artifact_keys: saugomų artefaktų raktus, kad juos būtų galima greitai patikrinti

`Queuety::retry_workflow( $workflow_id )`

Pakartotinai paleiskite nepavykusią darbo eigą nuo nepavykusio žingsnio.

PHP

Queuety::retry_workflow( $workflow_id );

Išmeta RuntimeException, jei darbo eiga po nesėkmės jau buvo kompensuota.

`Queuety::pause_workflow( $workflow_id )`

Pristabdykite vykdomą darbo eigą. Jau į eilę įtrauktas dabartinio žingsnio darbas vis dar gali būti užbaigtas, tačiau vėlesni žingsniai nebus įtraukti į eilę, kol neatnaujinsite darbo eigos.

PHP

Queuety::pause_workflow( $workflow_id );

`Queuety::resume_workflow( $workflow_id )`

Atnaujinkite pristabdytą darbo eigą.

PHP

Queuety::resume_workflow( $workflow_id );

`Queuety::cancel_workflow( $workflow_id )`

Atšaukite darbo eigą ir vykdykite visas užregistruotas išvalymo apdorojimo funkcijas. Būsena nustatoma į cancelled, o visi laukiantys ir vykdomi šios darbo eigos darbai perkeliami į archyvą.

PHP

Queuety::cancel_workflow( $workflow_id );

Išmeta RuntimeException, jei darbo eiga jau yra completed arba cancelled. Išsamiau žr. Atšaukimas.

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

Grąžinkite darbo eigą į ankstesnį žingsnį ir paleiskite ją iš naujo nuo ten. Nurodyto žingsnio būsena atkuriama iš įvykių žurnalo state_after, o kitas žingsnis įtraukiamas į eilę.

PHP

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

Išmeta RuntimeException, jei tiksliniam žingsniui nėra įrašytos būsenos. Reikalingas Darbo eigos įvykių žurnalas. Išsamiau žr. Kelionė laiku.

`Queuety::fork_workflow( $workflow_id )`

Išskaidykite vykdomą darbo eigą į nepriklausomą kopiją su dabartine jos būsena. Grąžinamas naujos darbo eigos ID.

PHP

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

Išskaidyta darbo eiga gauna atskirą ID ir veikia nepriklausomai nuo pradinės. Išsamiau žr. Išskaidymas.

`Queuety::export_workflow( $workflow_id )`

Eksportuokite visą darbo eigos vykdymo istoriją kaip JSON suderinamą masyvą. Įtraukiama darbo eigos eilutė, visi darbai, darbo eigos įvykiai ir žurnalo įrašai.

PHP

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

Apie JSON struktūrą ir panaudojimo atvejus žr. Eksportavimas ir atkūrimas.

`Queuety::replay_workflow( $data )`

Atkurkite eksportuotą darbo eigą dabartinėje aplinkoje. Iš eksporto duomenų sukuriama nauja darbo eiga ir grąžinamas jos ID.

PHP

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

Išsamiau žr. Eksportavimas ir atkūrimas.

`Queuety::replace_wp_cron()`

Pakeiskite įtaisytąją WordPress cron sistemą Queuety planuokliu. Perimamos wp_schedule_event(), wp_unschedule_event() ir wp_get_scheduled_event(), o pasikartojantys įvykiai nukreipiami per Queuety darbininką.

PHP

Queuety::replace_wp_cron();

Veikia tik tada, kai įkeltas WordPress. Išsamiau žr. WP-Cron pakeitimas.

`Queuety::restore_wp_cron()`

Atkurkite numatytąją WordPress cron elgseną pašalindami Queuety cron integravimo kablius.

PHP

Queuety::restore_wp_cron();

Apribojimus rasite skyriuje WP-Cron pakeitimas.

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

Išsiųskite signalą darbo eigai. Jei darbo eiga šiuo metu laukia šio signalo, ji iškart tęsiama. Kitu atveju signalas išsaugomas ir apdorojamas, kai darbo eiga pasiekia atitinkamą wait_for_signal žingsnį.

PHP

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

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

Išsamiau žr. Darbo eigos signalai.

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

Semantinė signal() apvalkalo funkcija patvirtinimo eigoms.

PHP

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

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

Semantinė signal() apvalkalo funkcija atmetimo eigoms.

PHP

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

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

Semantinė signal() apvalkalo funkcija struktūruotai žmogaus įvesčiai.

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

Išsaugokite vieną patvarų darbo eigos artefaktą.

PHP

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

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

Išsaugokite vieną šiuo metu vykdomo darbo eigos žingsnio artefaktą.

PHP

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

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

Gaukite vieną išsaugotą darbo eigos artefaktą.

PHP

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

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

Išvardykite darbo eigos artefaktus.

PHP

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

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

Ištrinkite vieną darbo eigos artefaktą.

PHP

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

Būsenų automatai

`Queuety::machine( $name )`

Pradėkite kurti būsenų automatą. Grąžinamas StateMachineBuilder.

PHP

use Queuety\Enums\StateMachineStatus;

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

`StateMachineBuilder` metodai

Metodas	Aprašas
`initial( string $state_name )`	Aiškiai nustatykite pradinę būseną.
`state( string $state_name, ?StateMachineStatus $terminal_status = null )`	Pradėkite konfigūruoti vieną būseną.
`action( string\|array $action_class )`	Užregistruokite į eilę įtraukiamą dabartinės būsenos įėjimo veiksmą. Nurodykite klasės pavadinimą arba struktūruotą apibrėžtį su `class`, `payload` ir `config`; `payload` pasiekia `StateAction::handle()` kaip `$payload`.
`on( string $event, string $target, string\|array\|null $guard_class = null, ?string $name = null )`	Užregistruokite vieną įvykio perėjimą iš dabartinės būsenos. Apsaugos sąlyga priima tą patį struktūruotą formatą kaip `action()`.
`on_queue( string $queue )`	Nustatykite eilę, naudojamą į eilę įtrauktiems būsenos veiksmams.
`with_priority( Priority $priority )`	Nustatykite į eilę įtrauktų būsenos veiksmų prioritetą.
`max_attempts( int $max_attempts )`	Nustatykite į eilę įtrauktų būsenos veiksmų pakartotinių bandymų ribą.
`version( string $version )`	Pažymėkite automato apibrėžtį programos lygmens versija.
`idempotency_key( string $key )`	Užtikrinkite patvarų išsiuntimą pagal skambintojo pateiktą raktą. Pakartotinai išsiuntus tą patį raktą, grąžinamas pradinis automato ID.
`dispatch( array $initial_state = [] )`	Išsiųskite automatą ir grąžinkite jo ID.

Būsenos įėjimo veiksmų klasės taip pat gali turėti pasirenkamą metodą config() su tais pačiais išteklių strategijos raktais, kuriuos palaiko darbo eigos žingsniai: concurrency_group, concurrency_limit ir cost_units.

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

Išsiųskite nuosekliai išsaugotą būsenų automato apibrėžties paketą. Daugumoje programos kodo vietų verčiau naudokite machine()->dispatch(). Šis žemesnio lygio metodas skirtas adapteriams, kurie paleidžia automatą tiesiogiai nepriklausydami nuo kūrimo priemonės.

PHP

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

`Queuety::machine_status( $machine_id )`

Gaukite dabartinę automato būseną. Grąžinamas objektas StateMachineState arba 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 taip pat pateikia:

definition_version
definition_hash
idempotency_key
error_message
current_action
terminal_status

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

Išsiųskite vieną išorinį įvykį laukiančiam automatui.

PHP

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

Įkrova prieinama perėjimų apsaugos sąlygoms ir, įvykį priėmus, sujungiama su viešąja automato būsena.

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

Išvardykite išsaugotas automato eilutes, kad galėtumėte jas patikrinti.

PHP

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

Kiekvienoje eilutėje pateikiamas automato ID, pavadinimas, būsena, dabartinė būsena, apibrėžties versija, apibrėžties maiša, idempotentiškumo raktas ir gyvavimo ciklo laiko žymos.

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

Gaukite vieno automato patvarią įvykių laiko juostą.

PHP

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

Kiekvienoje įvykio eilutėje yra: 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 ir created_at.

`Queuety::machine_trace( $machine_id )`

Gaukite normalizuotą pėdsako paketą būsenų automato derintuvo UI.

PHP

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

Pėdsake yra machine, sugrupuoti states, neapdoroti events, koreliuotas veiksmas jobs, logs, artifacts ir chunks.

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

Gaukite viešąją būseną po konkretaus automato pėdsako įvykio arba null, jei nerasta.

PHP

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

Apie gyvavimo ciklo modelį ir naudojimo šablonus žr. Būsenų automatai.

Darbo eigos šablonai

`Queuety::define_workflow( $name )`

Sukurkite darbo eigos kūrimo priemonę, kurią naudosite kaip šabloną.

PHP

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

`Queuety::register_workflow_template( $builder )`

Užregistruokite darbo eigos šabloną.

PHP

Queuety::register_workflow_template( $builder );

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

Išsiųskite užregistruotą darbo eigos šabloną pagal pavadinimą. Grąžinamas darbo eigos ID.

PHP

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

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

Išmeta RuntimeException, jei šablonas neužregistruotas.

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

Išsiųskite darbo eigą iš vykdymo metu pateikto apibrėžties paketo.

Daugumoje programos kodo vietų verčiau naudokite run_workflow() arba workflow()->dispatch(). Šis žemesnio lygio metodas skirtas daugkartinio naudojimo adapteriams, kurie išsiunčia nuosekliai išsaugotą darbo eigos apibrėžtį tiesiogiai nepriklausydami nuo vidinių darbo eigos tvarkytuvo metodų.

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

Užregistruokite WordPress veiksmą, kuris išsiunčia darbo eigą.

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 gali būti:

užregistruoto darbo eigos šablono pavadinimas
tiesioginė WorkflowBuilder

map, when ir idempotency_key vykdomi sinchroniškai dabartinėje WordPress užklausoje. Tik darbo eigos žingsniai vykdomi asinchroniškai po išsiuntimo.

Visą vadovą rasite skyriuje Veiksmų aktyvikliai.

`Queuety::workflow_templates()`

Gaukite darbo eigos šablonų registrą.

PHP

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

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

Įkelkite ir užregistruokite visus darbo eigos failus iš katalogo. Kiekvienas .php failas turi grąžinti WorkflowBuilder. Grąžinamas užregistruotų darbo eigų skaičius.

PHP

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

`Queuety::load_workflow_file( $file_path )`

Įkelkite ir užregistruokite vieną darbo eigos failą. Grąžinama WorkflowTemplate arba null, jei failas negrąžina kūrimo priemonės.

PHP

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

Planavimas

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

Sukurkite pasikartojantį tvarkaraštį. Grąžinama PendingSchedule kūrimo priemonė.

PHP

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

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

`PendingSchedule` metodai

Metodas	Aprašas
`every( string $interval )`	Nustatykite intervalo išraišką (e.g. `1 hour`, `30 minutes`)
`cron( string $expression )`	Nustatykite 5 laukų cron išraišką
`overlap( OverlapPolicy $policy )`	Nustatykite persidengimo strategiją (numatytoji: `Allow`). Žr. Planavimas.
`on_queue( string $queue )`	Nustatykite tikslinę eilę (numatytoji: `default`)
`id()`	Priverstinai išsiųskite ir grąžinkite tvarkaraščio ID

`Queuety::scheduler()`

Gaukite vidinį Scheduler egzempliorių.

PHP

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

Apdorojimo funkcijų registravimas

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

Užregistruokite apdorojimo klasę nurodytu pavadinimu.

PHP

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

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

Automatiškai aptikite ir užregistruokite apdorojimo klases iš katalogo. Grąžinamas užregistruotų apdorojimo funkcijų skaičius.

PHP

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

`Queuety::registry()`

Gaukite apdorojimo funkcijų registrą.

PHP

$registry = Queuety::registry();

Stebimumas

`Queuety::logger()`

Gaukite Logger egzempliorių, skirtą žurnalo įrašams užklausti.

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

Gaukite Metrics egzempliorių.

PHP

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

`Queuety::webhook_notifier()`

Gaukite WebhookNotifier egzempliorių.

PHP

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

`Queuety::rate_limiter()`

Gaukite RateLimiter egzempliorių.

PHP

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

Spartinančioji atmintinė

`Queuety::cache()`

Gaukite aktyvų spartinančiosios atmintinės posistemės egzempliorių.

PHP

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

`Queuety::set_cache( $cache )`

Pakeiskite spartinančiosios atmintinės posistemę. Iškvieskite prieš Queuety::init(), kad galėtumėte naudoti pasirinktinį įgyvendinimą.

PHP

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

`QUEUETY_CACHE_TTL`

Numatytoji spartinančiosios atmintinės galiojimo trukmė sekundėmis (numatytoji: 5). Nustatykite šią konstantą faile wp-config.php, kad pakeistumėte numatytąją vidinių spartinančiosios atmintinės operacijų galiojimo trukmę.

PHP

define( 'QUEUETY_CACHE_TTL', 10 );

Išsamiau apie posistemes ir automatinį aptikimą žr. Spartinimas atmintyje.

Būsenų automato įvykių žurnalas

`Queuety::machine_events()`

Gaukite StateMachineEventLog egzempliorių.

PHP

$log = Queuety::machine_events();

Darbo eigos įvykių žurnalas

`Queuety::workflow_events()`

Gaukite WorkflowEventLog egzempliorių.

PHP

$log = Queuety::workflow_events();

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

Gaukite visą darbo eigos įvykių laiko juostą. Grąžinamas įvykių eilučių masyvas, surikiuotas pagal ID.

PHP

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

Kiekvienoje įvykio eilutėje yra: 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 (nukopijuota iš error.message), duration_ms ir created_at.

`Queuety::workflow_trace( $workflow_id )`

Gaukite normalizuotą pėdsako paketą darbo eigos derintuvo UI.

PHP

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

Pėdsake yra workflow, sugrupuoti steps, neapdoroti events, jobs, logs, artifacts, chunks, signals ir wait_dependencies.

`Queuety::trace_input( $input )`

Nustatykite programos lygmens įvestį šiuo metu vykdomam darbo eigos žingsniui.

`Queuety::trace_output( $output )`

Nustatykite programos lygmens išvestį šiuo metu vykdomam darbo eigos žingsniui.

`Queuety::trace_context( array $context )`

Sujunkite programos lygmens kontekstą su šiuo metu vykdomu darbo eigos žingsniu.

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

Gaukite viešąją būseną, buvusią po to, kai konkretus darbo eigos žingsnis buvo užbaigtas arba pratęstas, arba null, jei nerasta.

PHP

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

Panaudojimo atvejus ir CLI komandas rasite skyriuje Darbo eigos įvykių žurnalas.

Vidiniai egzemplioriai

Šie metodai grąžina vidinius egzempliorius, skirtus pažangesniems naudojimo atvejams.

Metodas	Grąžinama reikšmė	Aprašas
`Queuety::queue()`	`Queue`	Eilės operacijos (paėmimas, atlaisvinimas, perkėlimas į archyvą)
`Queuety::worker()`	`Worker`	Darbininko proceso valdymas
`Queuety::workflow_manager()`	`Workflow`	Darbo eigos koordinavimas
`Queuety::machines()`	`StateMachine`	Būsenų automato gyvavimo ciklo valdymas
`Queuety::connection()`	`Connection`	Duomenų bazės ryšys
`Queuety::batch_manager()`	`BatchManager`	Paketų gyvavimo ciklo valdymas (atšaukimas, valymas)
`Queuety::chunk_store()`	`ChunkStore`	Srautinių žingsnių dalių išsaugojimas
`Queuety::machine_events()`	`StateMachineEventLog`	Būsenų automato pėdsako įvykiai ir normalizuoti pėdsakų paketai
`Queuety::workflow_events()`	`WorkflowEventLog`	Darbo eigos pėdsako įvykiai ir normalizuoti pėdsakų paketai
`Queuety::resource_manager()`	`ResourceManager`	Darbininkų priėmimo profiliai, svertiniai biudžetai ir sistemos atminties tikrinimas
`Queuety::cache()`	`Cache`	Spartinančiosios atmintinės posistemės egzempliorius

Testavimas

`Queuety::fake()`

Pakeiskite eilę atmintyje veikiančiu pakaitalu, skirtu testavimui. Grąžinamas QueueFake egzempliorius, kuris įrašo išsiųstas užduotis neliesdamas duomenų bazės. create_batch() taip pat įrašo fiktyvius paketų metaduomenis; grandinės tikrinamos pagal atskiras užduotis, kurias jos išsiunčia.

PHP

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

`Queuety::reset()`

Nustatykite iš naujo visą vidinę būseną, įskaitant eilės pakaitalą. Iškvieskite per tearDown().

PHP

Queuety::reset();

`QueueFake` metodai

Metodas	Aprašas
`assert_pushed( string $class, ?Closure $callback )`	Patikrinkite, ar buvo įtraukta klasės užduotis. Pasirenkamas atgalinis iškvietimas filtruoja pagal užduoties duomenis.
`assert_pushed_times( string $class, int $count )`	Patikrinkite, ar užduotis buvo įtraukta tiksliai N kartų.
`assert_not_pushed( string $class )`	Patikrinkite, ar užduotis niekada nebuvo įtraukta.
`assert_nothing_pushed()`	Patikrinkite, ar apskritai nebuvo įtraukta jokių užduočių.
`assert_batched( ?Closure $callback )`	Patikrinkite, ar paketas buvo išsiųstas. Pasirenkamas atgalinis iškvietimas filtruoja pagal paketo duomenis.
`pushed( string $class )`	Gaukite visus įrašytus konkrečios klasės išsiuntimus. Grąžinamas masyvas.
`batches()`	Gaukite visus įrašytus paketus. Grąžinamas masyvas.
`reset()`	Išvalykite visus įrašytus darbus ir paketus.

Išsamius pavyzdžius rasite skyriuje Testavimas.

Sąsajos

`Handler`

Paprastų užduočių apdorojimo funkcijų sąsaja.

PHP

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

config() gali grąžinti queue, max_attempts, backoff, rate_limit, concurrency_group, concurrency_limit ir cost_units.

`Step`

Darbo eigos žingsnių apdorojimo funkcijų sąsaja.

PHP

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

config() gali grąžinti max_attempts, backoff, concurrency_group, concurrency_limit ir cost_units.

`Contracts\ForEachHandler`

Dinaminio šakos apdorojimo kiekvienam elementui funkcijų sąsaja.

PHP

namespace Queuety\Contracts;

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

config() gali grąžinti max_attempts, backoff, rate_limit, concurrency_group, concurrency_limit ir cost_units.

`Contracts\ForEachReducer`

Pasirenkamas for-each rezultatų reduktorius.

PHP

namespace Queuety\Contracts;

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

`Contracts\Compensation`

Žingsnių kompensavimo apdorojimo sąsaja.

PHP

namespace Queuety\Contracts;

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

`Contracts\RepeatCondition`

Ciklo sąlygų sąsaja, naudojama repeat_until() ir repeat_while(), kai būsenos rakto palyginimo nepakanka.

PHP

namespace Queuety\Contracts;

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

`Contracts\StateAction`

Į eilę įtraukiamų būsenos veiksmų sąsaja.

PHP

namespace Queuety\Contracts;

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

$payload perduoda struktūrizuotą payload iš veiksmo aprašo, jei mašina buvo apibrėžta su juo. Grąžinkite viešus būsenos atnaujinimus. Įtraukite _event ir, jei reikia, _event_payload, kai veiksmas turi iškart pereiti į kitą būseną.

Būsenos veiksmai taip pat gali apibrėžti pasirenkamą metodą config(). Queuety nuskaito iš jo concurrency_group, concurrency_limit ir cost_units, kai į eilę įtraukiamas įėjimo veiksmas.

`Contracts\StateGuard`

Būsenų mašinos perėjimų apsaugų sąsaja.

PHP

namespace Queuety\Contracts;

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

`Contracts\Job`

Savarankiškų iškviečiamų užduočių klasių sąsaja. Užduočių klasės įkapsuliuoja duomenis viešosiose savybėse, o vykdymo logiką – handle().

PHP

namespace Queuety\Contracts;

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

Užduočių klasės gali pasirinktinai apibrėžti metodą middleware(), grąžinantį Contracts\Middleware egzempliorių masyvą.

`Contracts\Cache`

Talpyklos vidinių realizacijų sąsaja. Žr. Talpyklos, kur rasite integruotas realizacijas ir pasirinktines vidines realizacijas.

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

Metodas	Grąžina	Aprašymas
`get( $key )`	`mixed`	Gauti reikšmę. Jei ji nerasta arba jos galiojimo laikas pasibaigęs, grąžinama `null`.
`set( $key, $value, $ttl )`	`void`	Išsaugoti reikšmę. TTL reikšmė 0 reiškia, kad galiojimo laikas neribojamas.
`delete( $key )`	`void`	Pašalinti reikšmę.
`has( $key )`	`bool`	Patikrinti, ar raktas egzistuoja ir jo galiojimo laikas nepasibaigęs.
`add( $key, $value, $ttl )`	`bool`	Atominis nustatymas, jei reikšmės nėra. Grąžina `true`, jei reikšmė nustatyta.
`flush()`	`void`	Pašalinti visus elementus iš talpyklos.

`WorkflowEventLog`

Registruoja darbo eigos žingsnių perėjimus kartu su visu sekimo kontekstu. Pasiekiama per Queuety::workflow_events().

Metodas	Grąžina	Aprašymas
`record_event( array $event )`	`void`	Įrašyti neapdorotą sekimo įvykį
`record_step_started( ... )`	`void`	Įrašyti step_started įvykį su įvestimi ir būsena prieš vykdymą
`record_step_completed( ... )`	`void`	Įrašyti step_completed įvykį su įvestimi, išvestimi, būsena prieš vykdymą ir po jo
`record_step_failed( ... )`	`void`	Įrašyti step_failed įvykį su sekimo kontekstu ir struktūrizuota klaidos informacija
`get_timeline( int $workflow_id, ?int $limit = 100, int $offset = 0 )`	`array`	Gauti darbo eigos sekimo įvykius eilės tvarka, pasirinktinai puslapiuojant
`get_trace( int $workflow_id )`	`array`	Gauti normalizuotą sekimo rinkinį
`get_state_at_step( int $workflow_id, int $step_index )`	`array\|null`	Gauti viešąją būseną po konkretaus žingsnio užbaigimo arba tęsimo
`prune( int $older_than_days )`	`int`	Ištrinti įvykius, senesnius už nurodytą dienų skaičių. Grąžina ištrintų eilučių skaičių.

`StateMachineEventLog`

Registruoja būsenų mašinos gyvavimo ciklo, veiksmų, perėjimų ir apsaugų įvykius, naudodama tuos pačius sekimo laukus kaip darbo eigos įvykiuose.

Metodas	Grąžina	Aprašymas
`record_event( array $event )`	`void`	Įrašyti neapdorotą mašinos sekimo įvykį
`get_timeline( int $machine_id, ?int $limit = 100, int $offset = 0 )`	`array`	Gauti mašinos sekimo įvykius eilės tvarka, pasirinktinai puslapiuojant
`get_trace( int $machine_id )`	`array`	Gauti normalizuotą mašinos sekimo rinkinį
`get_state_at_event( int $machine_id, int $event_id )`	`array\|null`	Gauti viešąją būseną po konkretaus mašinos sekimo įvykio
`prune( int $older_than_days )`	`int`	Ištrinti įvykius, senesnius už nurodytą dienų skaičių. Grąžina ištrintų eilučių skaičių.

`Contracts\StreamingStep`

Darbo eigos žingsnių apdorojimo sąsaja, generuojanti duomenų dalių srautą. Kiekviena generuojama reikšmė iškart išsaugoma duomenų bazėje. Žr. Srautiniai žingsniai.

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

Metodas	Aprašymas
`stream( $state, $existing_chunks )`	Generatorius, generuojantis tekstines dalis. Kiekviena generuojama dalis iškart įrašoma į duomenų bazę. Bandant iš naujo, `$existing_chunks` yra anksčiau išsaugotos dalys.
`on_complete( $chunks, $state )`	Iškviečiamas srautui pasibaigus. Grąžina duomenis, kuriuos reikia sujungti su darbo eigos būsena.
`config()`	Pasirenkama konfigūracija. Palaiko `max_attempts`, `backoff`, `concurrency_group`, `concurrency_limit` ir `cost_units`.

`ChunkStore`

Tvarko srautinio perdavimo žingsnių dalių išsaugojimą. Pasiekiama per Queuety::chunk_store().

Metodas	Grąžina	Aprašymas
`get_chunks( int $job_id )`	`string[]`	Gauti visas užduoties dalis, surikiuotas pagal dalies indeksą
`append_chunk( int $job_id, int $chunk_index, string $content, ?int $workflow_id, ?int $step_index )`	`void`	Pridėti vieną dalį
`clear_chunks( int $job_id )`	`void`	Ištrinti visas užduoties dalis
`chunk_count( int $job_id )`	`int`	Suskaičiuoti užduoties dalis
`get_accumulated( int $job_id )`	`string`	Sujungti visas dalis į vieną eilutę

`Heartbeat`

Statinis pagalbinis įrankis veiklos pulsams siųsti iš žingsnių apdorojimo ir srautinio perdavimo žingsnių. Žr. Pulsai.

Metodas	Aprašymas
`Heartbeat::init( int $job_id, Connection $conn )`	Nustatyti užduoties kontekstą. Darbuotojas iškviečia automatiškai.
`Heartbeat::beat( array $progress = [] )`	Siųsti pulsą. Atnaujina `reserved_at` ir, pasirinktinai, išsaugo eigos duomenis `heartbeat_data`.
`Heartbeat::clear()`	Išvalyti užduoties kontekstą. Darbuotojas iškviečia automatiškai.
`Heartbeat::current_job_id()`	Gauti dabartinės užduoties ID (testavimui).

`Contracts\Middleware`

Užduočių tarpinės programinės įrangos sąsaja. Tarpinė programinė įranga apgaubia užduoties vykdymą svogūno sluoksnių tipo grandine.

PHP

namespace Queuety\Contracts;

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

Savybių rinkiniai

`Dispatchable`

Suteikia statinius iškvietimo metodus užduočių klasėms, įgyvendinančioms Contracts\Job. Konstruktoriaus argumentai perduodami užduoties egzemplioriui sukurti.

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

Metodas	Grąžina	Aprašymas
`dispatch( ...$args )`	`PendingJob`	Sukurti egzempliorių ir išsiųsti jį vykdyti
`dispatch_if( bool $condition, ...$args )`	`PendingJob\|null`	Išsiųsti vykdyti tik tada, jei `$condition` yra true
`dispatch_unless( bool $condition, ...$args )`	`PendingJob\|null`	Išsiųsti vykdyti tik tada, jei `$condition` yra false
`with_chain( array $jobs )`	`ChainBuilder`	Pradėti grandinę nuo šios užduoties klasės

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

Užduočių savybės

Užduočių klasės gali deklaruoti viešąsias savybes, kurias darbuotojas nuskaito pasitelkęs refleksiją, kad pakeistų pakartotinių bandymų ir skirtojo laiko elgseną.

Savybė	Tipas	Aprašymas
`$tries`	`int`	Didžiausias bandymų skaičius prieš užduotį perkeliant į archyvą
`$timeout`	`int`	Didžiausia vykdymo trukmė sekundėmis (reikalingas `pcntl`)
`$backoff`	`array`	Didėjančios delsos reikšmės sekundėmis kiekvienam pakartotiniam bandymui

PHP

class ImportProductsJob implements Job {
    use Dispatchable;

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

    // ...
}

`failed()` kabliukas

Kai užduotis perkeliama į archyvą išnaudojus visus pakartotinius bandymus, darbuotojas iškviečia užduoties egzemplioriaus metodą failed(), jei jis apibrėžtas.

PHP

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

Integruota tarpinė programinė įranga

Visos tarpinės programinės įrangos klasės įgyvendina Contracts\Middleware ir yra Queuety\Middleware vardų srityje.

Klasė	Aprašymas
`RateLimited( int $max, int $window )`	Užtikrinti, kad per `$window` sekundžių būtų įvykdyta ne daugiau kaip `$max` vykdymų
`Timeout( int $seconds )`	Nutraukti užduotį po `$seconds` sekundžių (reikalingas `pcntl`)
`UniqueJob( ?string $key )`	Duomenų bazės užraktu neleisti vykdyti vienu metu
`WithoutOverlapping( string $key, int $release_after )`	Neleisti vykdyti persidengiančių užduočių, automatiškai pasibaigiant užrakto galiojimui
`ThrottlesExceptions( int $max_attempts, int $decay_minutes )`	Riboti išimčių dažnį, kad išorinei paslaugai neveikiant nekiltų užduočių antplūdis

Naudojimo informaciją ir pavyzdžius rasite skiltyje Tarpinė programinė įranga.

Enum tipai

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

Informacijos apie kiekvienos strategijos veikimą rasite skiltyje Planavimas.

Atributai

`#[QueuetyHandler]`

Automatinės registracijos 8 atributas PHP. Žr. PHP atributus.

PHP

use Queuety\Attributes\QueuetyHandler;

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

CLI tiltas

`Queuety::cli_command()`

Grąžina paskelbtos WP-CLI šakninės komandos pavadinimą.

Numatytoji šakninė komanda yra queuety. Apibrėžkite QUEUETY_CLI_COMMAND prieš paleisdami Queuety, jei komandą norite paskelbti kitu pavadinimu.

`Queuety::cli_command_map()`

Grąžina serializuojamą Queuety CLI katalogą, kad testavimo aplinkoms būtų prieinamas pristatytų komandų aprašas ir nereikėtų nuskaityti WP-CLI klasių.

Kiekviename įraše pateikiama:

pastovus operacijos ID
santykinis CLI kelias
WP-CLI apdorojimo klasė ir metodas
galimi vykdymo tikslai
adapteris, normalizuojantis neapdorotus CLI argumentus į vykdymo planą

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

Paverčia iškviestą išanalizuotą CLI vykdymo planu, kurį turėtų vykdyti testavimo aplinka.

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

Vykdymo planuose atsižvelgiama į transportą:

komandos susiejamos su viešai prieinamais PHP iškviečiamais objektais ir normalizuotais argumentais
kataloge iš anksto išvardijami visi galimi vykdymo tikslai, o callable ir transport nurodo išnagrinėtos įvesties pasirinktą šaką

Tai standartinis būdas, kurį Queuety naudoja testavimo aplinkose agentams: pastovus operacijos ID, viešai prieinamas API tikslas ir adapterio atliekamas normalizavimas CLI specifiniams aspektams, pavyzdžiui, numatytosioms eilėms, JSON duomenų iškodavimui, darbuotojų telkinio šakojimui ir darbo eigos eksporto failų tvarkymui.

PHP API

Atskira prieiga prie pagrindinių komponentų

Queuety::init( Connection $conn )

Queuety::ensure_schema()

Užduotys

Queuety::dispatch( $handler, $payload )

PendingJob metodai

Queuety::dispatch_sync( $handler, $payload )

Queuety::batch( $jobs )

Queuety::create_batch( $jobs )

BatchBuilder metodai

Queuety::find_batch( $id )

Vykdytojų valdymas

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

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

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

Batch reikšmės objektas

Queuety::chain( $jobs )

ChainBuilder metodai

Queuety::retry( $job_id )

Queuety::retry_buried()

Queuety::purge( $older_than_days )

Eilės

Queuety::stats( $queue )

Queuety::buried( $queue )

Queuety::pause( $queue )

Queuety::resume( $queue )

Queuety::is_paused( $queue )

Darbo eigos

Queuety::workflow( $name )

WorkflowBuilder metodai

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 )