PHP API

Bekijken als Markdown Bewerken op GitHub Documentatie op 2cc7b2b

Alles in Queuety is toegankelijk via de statische klasse Queuety\Queuety. Dit is het enige toegangspunt voor het dispatchen van jobs, het bouwen van workflows, het aansturen van state machines, het beheren van queues en toegang tot interne subsystemen.

PHP

use Queuety\Queuety;

Toegang tot de zelfstandige kern

De DB-ondersteunde runtime van Queuety kan buiten WordPress worden geïnitialiseerd.

Deze zelfstandige route is bedoeld voor de orkestratiekern:

jobs en workers (workerpools hebben ook de constanten DB_HOST, DB_NAME, DB_USER en DB_PASSWORD nodig; zie hieronder)
workflows en signalen
state machines en events
artefacten, logs, metrics en planningen

Het is geen vervanging voor de WordPress-adapter. Queuety::on_action(), replace_wp_cron(), het opstarten van de plugin en het echte registratiepad voor wp queuety ... vereisen nog steeds WordPress.

`Queuety::init( Connection $conn )`

Initialiseer Queuety met een directe DB-verbinding.

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

Het optionele argument table_prefix wijzigt het Queuety-gedeelte van de runtimetabelnamen per verbinding:

PHP

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

Daardoor worden tabellen zoals queuety_jobs omgezet naar wp_themequeue_jobs.

`Queuety::ensure_schema()`

Zorg dat de runtimetabellen van Queuety bestaan voor de actieve verbinding.

PHP

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

Zie Zelfstandig gebruik voor het volledige model en de grens van WordPress.

Jobs

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

Dispatch een job. Accepteert een handlernaam als string of een instantie van Contracts\Job. Retourneert een PendingJob-builder om opties aan elkaar te koppelen.

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

Als een instantie van Contracts\Job wordt doorgegeven, haalt de serializer de FQCN op als handlernaam en de publieke properties als payload. De parameter $payload wordt genegeerd als $handler een instantie van Job is.

Methoden van `PendingJob`

Methode	Beschrijving
`on_queue( string $queue )`	Stel de doelqueue in
`with_priority( Priority $priority )`	Stel het prioriteitsniveau in
`delay( int $seconds )`	Vertraging voordat de job beschikbaar komt
`max_attempts( int $max )`	Maximaal aantal pogingen
`rate_limit( int $max, int $window )`	Snelheidslimiet: maximaal aantal uitvoeringen per tijdvenster in seconden
`unique()`	Voorkom dubbele dispatches
`after( int $job_id )`	Wacht eerst tot een andere job is voltooid
`concurrency_group( string $group, int $limit )`	Plaats de job in een gedeelde groep voor worker-toelating
`cost_units( int $units )`	Voeg metadata met de relatieve uitvoeringskosten toe aan de dispatch
`id()`	Forceer de dispatch en retourneer de job-ID

concurrency_group() en cost_units() maken deel uit van het bredere model voor Resourcebeheer. Zo kan dezelfde dispatch deelnemen aan limieten voor gelijktijdigheid op basis van aantallen, gewogen budgetten voor queues of providers en heuristieken voor worker-toelating.

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

Voer een job synchroon uit in het huidige proces, zonder deze aan de queue toe te voegen.

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

De methode handle() van de job wordt rechtstreeks aangeroepen. Er wordt geen databaserecord aangemaakt, er is geen worker bij betrokken en middleware wordt niet toegepast.

`Queuety::batch( $jobs )`

Dispatch meerdere jobs in één INSERT met meerdere rijen.

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

Maak een BatchBuilder om een groep jobs te dispatchen met callbacks en voortgangsregistratie. Zie Batchverwerking voor alle details.

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

Methoden van `BatchBuilder`

Methode	Retourneert	Beschrijving
`name( string $name )`	`self`	Stel een naam in voor de batch
`on_queue( string $queue )`	`self`	Stel de queue in voor alle jobs in de batch
`then( string $handler_class )`	`self`	Handler die wordt aangeroepen wanneer de batch succesvol is voltooid
`catch( string $handler_class )`	`self`	Handler die wordt aangeroepen wanneer de eerste job in de batch mislukt
`finally( string $handler_class )`	`self`	Handler die wordt aangeroepen wanneer de batch eindigt (succes of mislukking)
`allow_failures()`	`self`	Sta toe dat de callback `then` wordt uitgevoerd, ook bij fouten
`dispatch()`	`Batch`	Maak de batchrij en alle jobs aan; retourneert het waardeobject `Batch`

`Queuety::find_batch( $id )`

Zoek een batch op ID. Retourneert een waardeobject Batch of null.

PHP

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

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

Workerbeheer

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

Voer rechtstreeks één worker-loop uit.

PHP

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

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

Voer rechtstreeks een workerpool met vaste omvang uit. Vereist de extensies pcntl en posix. De pool leest de constanten WordPress, DB_HOST, DB_NAME, DB_USER en DB_PASSWORD voor de onderliggende verbindingen (alleen de tabelprefixen komen van de geïnitialiseerde verbinding). Definieer ze daarom wanneer je een pool buiten WordPress uitvoert (#5).

PHP

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

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

Voer rechtstreeks een adaptieve workerpool uit. Dezelfde vereisten als bij run_worker_pool().

PHP

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

Waardeobject `Batch`

Methode/property	Type	Beschrijving
`$id`	`int`	Batch-ID
`$name`	`string\|null`	Optionele batchnaam
`$total_jobs`	`int`	Totaal aantal jobs in de batch
`$pending_jobs`	`int`	Jobs die nog in behandeling zijn
`$failed_jobs`	`int`	Aantal mislukte jobs
`$failed_job_ids`	`array`	ID's van mislukte jobs
`$created_at`	`DateTimeImmutable`	Aanmaaktijd van de batch
`$finished_at`	`DateTimeImmutable\|null`	Eindtijd van de batch
`$cancelled_at`	`DateTimeImmutable\|null`	Annuleringstijd van de batch
`progress()`	`int`	Percentage voltooid (0-100)
`finished()`	`bool`	Of de batch is voltooid
`has_failures()`	`bool`	Of een job is mislukt
`cancelled()`	`bool`	Of de batch is geannuleerd

`Queuety::chain( $jobs )`

Maak een ChainBuilder voor sequentiële uitvoering van jobs. Elke job is afhankelijk van de succesvolle voltooiing van de vorige.

PHP

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

Methoden van `ChainBuilder`

Methode	Retourneert	Beschrijving
`on_queue( string $queue )`	`self`	Stel de queue in voor alle jobs in de keten
`catch( string $handler_class )`	`self`	Handler die wordt aangeroepen wanneer een job in de keten definitief in de wachtrij blijft staan
`dispatch()`	`int`	Maak alle jobs aan met sequentiële afhankelijkheden; retourneert de ID van de eerste job

`Queuety::retry( $job_id )`

Probeer een specifieke job opnieuw. Zet deze terug naar de status pending.

PHP

Queuety::retry( 42 );

`Queuety::retry_buried()`

Probeer alle vastgelopen jobs opnieuw. Retourneert het aantal opnieuw geprobeerde jobs.

PHP

$count = Queuety::retry_buried();

`Queuety::purge( $older_than_days )`

Verwijder voltooide jobs. Retourneert het aantal verwijderde jobs.

PHP

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

Queues

`Queuety::stats( $queue )`

Haal het aantal jobs per status op.

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

Haal alle vastgelopen jobs op.

PHP

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

`Queuety::pause( $queue )`

Pauzeer een wachtrij zodat workers deze overslaan.

PHP

Queuety::pause( 'emails' );

`Queuety::resume( $queue )`

Hervat een gepauzeerde wachtrij.

PHP

Queuety::resume( 'emails' );

`Queuety::is_paused( $queue )`

Controleer of een wachtrij is gepauzeerd.

PHP

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

Workflows

`Queuety::workflow( $name )`

Begin met het bouwen van een workflow. Geeft een WorkflowBuilder terug.

PHP

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

`WorkflowBuilder`-methoden

Methode	Beschrijving
`then( string $handler_class, ?string $name = null )`	Voeg een opeenvolgende stap toe. Optionele naam voor voorwaardelijke vertakkingen.
`parallel( array $handler_classes )`	Voeg een groep parallelle stappen toe.
`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 )`	Voeg een dynamische for-each-stap toe die voor elk item in de workflowstatus één vertakkingsjob uitbreidt.
`repeat_while( string $target_step, ?string $state_key = null, mixed $expected = true, ?string $name = null, ?string $condition_class = null, ?int $max_iterations = null )`	Spring terug naar een eerder benoemde stap zolang een openbare statussleutel of `RepeatCondition`-klasse overeenkomt.
`repeat_until( string $target_step, ?string $state_key = null, mixed $expected = true, ?string $name = null, ?string $condition_class = null, ?int $max_iterations = null )`	Spring terug naar een eerder benoemde stap totdat een openbare statussleutel of `RepeatCondition`-klasse overeenkomt.
`run_workflow( string $workflow_name, WorkflowBuilder $sub_builder )`	Voeg een run-workflow-stap toe.
`delay( int $seconds = 0, int $minutes = 0, int $hours = 0, int $days = 0 )`	Voeg een duurzame vertragingsstap toe. Alle tijdsduren worden opgeteld.
`wait_for_signal( string $name, ?string $result_key = null, ?string $step_name = null, array $match_payload = [], ?string $correlation_key = null )`	Voeg een stap toe die op een signaal wacht. Optionele `match_payload` en `correlation_key` laten je niet-gerelateerde gebeurtenissen negeren.
`wait_for_signals( array $signal_names, WaitMode $mode = WaitMode::All, ?string $result_key = null, ?string $name = null, array $match_payload = [], ?string $correlation_key = null )`	Wacht op alle of een van meerdere signalen, met optionele payloadmatching en correlatie.
`wait_for_approval( string $signal_name = 'approval', ?string $result_key = 'approval', ?string $name = null, array $match_payload = [], ?string $correlation_key = null )`	Wacht op een goedkeuringssignaal van een persoon.
`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 )`	Wacht op goedkeuring of afwijzing en sla een gestructureerde beslissingspayload op.
`wait_for_input( string $signal_name = 'input', ?string $result_key = 'input', ?string $name = null, array $match_payload = [], ?string $correlation_key = null )`	Wacht op een invoersignaal van een persoon.
`wait_for_workflow( int\|string $workflow, ?string $result_key = null, ?string $name = null )`	Wacht tot één workflow-ID, of een workflow-ID die in de openbare status is opgeslagen, is voltooid.
`wait_for_workflows( array\|string $workflows, WaitMode $mode = WaitMode::All, ?string $result_key = null, ?string $name = null, ?int $quorum = null )`	Wacht tot alle, een willekeurig aantal of een quorum van meerdere workflows is voltooid.
`wait_for_workflow_group( string $group_key, WaitMode $mode = WaitMode::All, ?int $quorum = null, ?string $result_key = null, ?string $name = null )`	Wacht op een eerder gestarte workflowgroep via de interne groepssleutel.
`start_workflows( string $items_key, WorkflowBuilder $workflow_builder, string $result_key = 'started_workflow_ids', string $payload_key = 'item', bool $inherit_state = true, ?string $name = null, ?string $group_key = null )`	Start één workflow op hoofdniveau per runtime-item en geef de gestarte groep optioneel een duurzame groepssleutel.
`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 )`	Op agents gericht alias voor `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 )`	Op agents gericht alias voor `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 )`	Op agents gericht alias voor `wait_for_workflow_group()`.
`version( string $version )`	Voorzie de workflowdefinitie van een applicatieversie voor inspectie en exports.
`idempotency_key( string $key )`	Maak het starten van een workflow duurzaam met een door de aanroeper opgegeven sleutel. Opnieuw starten met dezelfde sleutel geeft de oorspronkelijke workflow-ID terug.
`max_transitions( int $max )`	Laat de workflow mislukken als er meer dan `$max` stapovergangen zijn voltooid.
`max_for_each_items( int $max )`	Laat de workflow mislukken als een `for_each()`-stap meer dan `$max` items probeert uit te breiden.
`max_state_bytes( int $bytes )`	Laat de workflow mislukken als de openbare status groter wordt dan `$bytes`.
`max_cost_units( int $max )`	Laat de workflow mislukken als voltooide stappen meer dan `$max` kosteneenheden verbruiken.
`max_started_workflows( int $max )`	Laat de workflow mislukken voordat er meer dan `$max` onderliggende workflows worden gestart.
`compensate_with( string $handler_class )`	Koppel een compensatiehandler aan de laatst toegevoegde stap.
`compensate_on_failure()`	Voer compensaties voor voltooide stappen automatisch uit als de workflow mislukt.
`on_queue( string $queue )`	Stel de wachtrij in voor alle stappen.
`with_priority( Priority $priority )`	Stel de prioriteit in voor alle stappen.
`max_attempts( int $max )`	Stel het maximale aantal nieuwe pogingen per stap in.
`on_cancel( string $handler_class )`	Registreer een klasse voor opschoning bij annulering. Zie Annulering.
`prune_state_after( int $steps = 2 )`	Schakel automatisch opschonen van de status in. Verwijdert uitvoer van stappen die ouder is dan de drempel. Zie Status opschonen.
`must_complete_within( int $seconds = 0, int $minutes = 0, int $hours = 0, int $days = 0 )`	Stel een deadline voor de workflow in. Alle tijdsduren worden opgeteld. Zie Deadlines.
`on_deadline( string $handler_class )`	Registreer een handler die wordt aangeroepen wanneer de workflowdeadline wordt overschreden. Zie Deadlines.
`dispatch( array $initial_payload = [] )`	Start de workflow. Geeft de workflow-ID terug.

`Queuety::workflow_status( $workflow_id )`

Vraag de workflowstatus op. Geeft een WorkflowState-object of null terug.

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 biedt ook:

waiting_for: de signaalnamen of workflow-ID's die de uitvoering momenteel blokkeren
current_step_name: de geconfigureerde naam van de huidige stap, indien aanwezig
wait_mode: de huidige wachtmodus (all, any of quorum) tijdens blokkering
wait_details: aanvullende inspectiegegevens, zoals overeenkomende en resterende blokkades
definition_version: de waarde die is ingesteld door version(), indien aanwezig
definition_hash: een deterministische hash van de workflowdefinitie
idempotency_key: de duurzame startsleutel, indien aanwezig
budget: geconfigureerde limieten en huidig gebruik, als beveiligingslimieten zijn ingeschakeld
artifact_count: het aantal opgeslagen artefacten voor de workflow
artifact_keys: opgeslagen artefact-sleutels voor snelle inspectie

`Queuety::retry_workflow( $workflow_id )`

Probeer een mislukte workflow opnieuw vanaf de stap waarop deze is mislukt.

PHP

Queuety::retry_workflow( $workflow_id );

Geeft RuntimeException als de workflow na een fout al is gecompenseerd.

`Queuety::pause_workflow( $workflow_id )`

Pauzeer een actieve workflow. Werk dat al voor de huidige stap in de wachtrij staat, kan nog worden voltooid, maar latere stappen worden pas in de wachtrij geplaatst wanneer je de workflow hervat.

PHP

Queuety::pause_workflow( $workflow_id );

`Queuety::resume_workflow( $workflow_id )`

Hervat een gepauzeerde workflow.

PHP

Queuety::resume_workflow( $workflow_id );

`Queuety::cancel_workflow( $workflow_id )`

Annuleer een workflow en voer geregistreerde opschoningshandlers uit. Stelt de status in op cancelled en begraaft alle wachtende en verwerkende jobs voor de workflow.

PHP

Queuety::cancel_workflow( $workflow_id );

Geeft RuntimeException als de workflow al completed of cancelled is. Zie Annulering voor meer informatie.

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

Zet een workflow terug naar een eerdere stap en voer deze vanaf daar opnieuw uit. De status wordt hersteld vanuit het gebeurtenislogboek state_after voor de opgegeven stap en de volgende stap wordt in de wachtrij geplaatst.

PHP

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

Geeft RuntimeException als er geen opgeslagen status voor de doelstap bestaat. Vereist het Workflow-gebeurtenislogboek. Zie Tijdreizen voor meer informatie.

`Queuety::fork_workflow( $workflow_id )`

Splits een actieve workflow af naar een onafhankelijke kopie met de huidige status. Geeft de nieuwe workflow-ID terug.

PHP

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

De afgesplitste workflow krijgt een eigen ID en werkt onafhankelijk van het origineel. Zie Afsplitsen voor meer informatie.

`Queuety::export_workflow( $workflow_id )`

Exporteer de volledige uitvoeringsgeschiedenis van een workflow als een JSON-serialiseerbare array. Bevat de workflowrij, alle jobs, workflowgebeurtenissen en logboekvermeldingen.

PHP

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

Zie Exporteren en opnieuw afspelen voor de JSON-structuur en gebruiksscenario's.

`Queuety::replay_workflow( $data )`

Speel een geëxporteerde workflow opnieuw af in de huidige omgeving. Maakt een nieuwe workflow aan op basis van de exportgegevens en geeft de nieuwe workflow-ID terug.

PHP

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

Zie Exporteren en opnieuw afspelen voor meer informatie.

`Queuety::replace_wp_cron()`

Vervang het ingebouwde WordPress cron-systeem door de scheduler van Queuety. Vangt wp_schedule_event(), wp_unschedule_event() en wp_get_scheduled_event() op en leidt terugkerende gebeurtenissen door de Queuety-worker.

PHP

Queuety::replace_wp_cron();

Werkt alleen als WordPress is geladen. Zie WP-Cron-vervanging voor meer informatie.

`Queuety::restore_wp_cron()`

Herstel het standaardgedrag van WordPress cron door de bridgehooks van Queuety cron te verwijderen.

PHP

Queuety::restore_wp_cron();

Zie WP-Cron-vervanging voor beperkingen.

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

Stuur een signaal naar een workflow. Als de workflow momenteel op dit signaal wacht, wordt deze direct hervat. Anders wordt het signaal opgeslagen en opgehaald wanneer de workflow de bijbehorende wait_for_signal-stap bereikt.

PHP

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

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

Zie Workflowsignalen voor meer informatie.

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

Semantische wrapper rond signal() voor goedkeuringsstromen.

PHP

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

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

Semantische wrapper rond signal() voor afwijzingsstromen.

PHP

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

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

Semantische wrapper rond signal() voor gestructureerde menselijke invoer.

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

Sla één duurzaam artefact op voor een workflow.

PHP

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

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

Sla één artefact op voor de momenteel uitgevoerde workflowstap.

PHP

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

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

Haal één opgeslagen artefact op voor een workflow.

PHP

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

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

Geef artefacten voor een workflow weer.

PHP

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

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

Verwijder één artefact voor een workflow.

PHP

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

Statusmachines

`Queuety::machine( $name )`

Begin met het bouwen van een statusmachine. Geeft een StateMachineBuilder terug.

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

Methode	Beschrijving
`initial( string $state_name )`	Stel de beginstatus expliciet in.
`state( string $state_name, ?StateMachineStatus $terminal_status = null )`	Begin met het configureren van één status.
`action( string\|array $action_class )`	Registreer een actie die in de wachtrij wordt geplaatst bij het betreden van de huidige status. Geef een klassenaam op of een gestructureerde definitie met `class`, `payload` en `config`; de `payload` bereikt `StateAction::handle()` als `$payload`.
`on( string $event, string $target, string\|array\|null $guard_class = null, ?string $name = null )`	Registreer één gebeurtenisovergang vanuit de huidige status. De guard accepteert dezelfde gestructureerde vorm als `action()`.
`on_queue( string $queue )`	Stel de wachtrij in die wordt gebruikt voor statusacties in de wachtrij.
`with_priority( Priority $priority )`	Stel de prioriteit in voor statusacties in de wachtrij.
`max_attempts( int $max_attempts )`	Stel de limiet voor nieuwe pogingen in voor statusacties in de wachtrij.
`version( string $version )`	Voorzie de machinedefinitie van een applicatieversie.
`idempotency_key( string $key )`	Maak het starten duurzaam met een door de aanroeper opgegeven sleutel. Opnieuw starten met dezelfde sleutel geeft de oorspronkelijke machine-ID terug.
`dispatch( array $initial_state = [] )`	Start de machine en geef de machine-ID terug.

Klassen voor statusinvoeracties kunnen ook een optionele config()-methode aanbieden met dezelfde sleutels voor resourcebeleid die workflowstappen ondersteunen: concurrency_group, concurrency_limit en cost_units.

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

Start een geserialiseerde bundel met een statusmachinedefinitie. De meeste applicatiecode gebruikt bij voorkeur machine()->dispatch(). Deze methode op lager niveau bestaat zodat adapters een machine kunnen starten zonder rechtstreeks afhankelijk te zijn van de builder.

PHP

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

`Queuety::machine_status( $machine_id )`

Vraag de huidige machinestatus op. Geeft een StateMachineState-object of null terug.

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 biedt ook:

definition_version
definition_hash
idempotency_key
error_message
current_action
terminal_status

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

Stuur één externe gebeurtenis naar een wachtende machine.

PHP

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

De payload is beschikbaar voor overgangscontroles en wordt samengevoegd met de openbare machinestatus wanneer de gebeurtenis wordt geaccepteerd.

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

Geef opgeslagen machinerijen weer ter inspectie.

PHP

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

Elke rij bevat de machine-id, naam, status, huidige status, definitieversie, definitiehash, idempotentiesleutel en tijdstempels van de levenscyclus.

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

Haal de duurzame gebeurtenistijdlijn voor één machine op.

PHP

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

Elke gebeurtenisrij bevat: 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 en created_at.

`Queuety::machine_trace( $machine_id )`

Haal een genormaliseerde tracebundel op voor een debugger voor statusmachines UI.

PHP

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

De trace bevat machine, gegroepeerde states, ruwe events, gecorreleerde actie-jobs, logs, artifacts en chunks.

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

Haal de openbare status op na een specifieke tracegebeurtenis van de machine, of null als deze niet is gevonden.

PHP

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

Zie Statusmachines voor het levenscyclusmodel en gebruikspatronen.

Workflowsjablonen

`Queuety::define_workflow( $name )`

Maak een workflowbuilder voor gebruik als sjabloon.

PHP

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

`Queuety::register_workflow_template( $builder )`

Registreer een workflowsjabloon.

PHP

Queuety::register_workflow_template( $builder );

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

Start een geregistreerd workflowsjabloon op naam. Geeft de workflow-ID terug.

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

Geeft RuntimeException als het sjabloon niet is geregistreerd.

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

Start een workflow vanuit een definitiebundel tijdens runtime.

De meeste applicatiecode gebruikt bij voorkeur run_workflow() of workflow()->dispatch(). Deze methode op lager niveau bestaat zodat herbruikbare adapters een geserialiseerde workflowdefinitie kunnen starten zonder afhankelijk te zijn van interne methoden van de workflowmanager.

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

Registreer een WordPress-actie die een workflow start.

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 kan het volgende zijn:

een geregistreerde naam van een workflowsjabloon
een inline WorkflowBuilder

map, when en idempotency_key worden synchroon uitgevoerd binnen het huidige WordPress-verzoek. Alleen de workflowstappen zelf worden na het starten asynchroon uitgevoerd.

Zie Actietriggers voor de volledige handleiding.

`Queuety::workflow_templates()`

Haal het register met workflowsjablonen op.

PHP

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

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

Laad en registreer alle workflowbestanden uit een map. Elk .php-bestand moet een WorkflowBuilder teruggeven. Geeft het aantal geregistreerde workflows terug.

PHP

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

`Queuety::load_workflow_file( $file_path )`

Laad en registreer één workflowbestand. Geeft de WorkflowTemplate terug, of null als het bestand geen builder teruggeeft.

PHP

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

Planning

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

Maak een terugkerende planning. Geeft een PendingSchedule-builder terug.

PHP

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

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

`PendingSchedule`-methoden

Methode	Beschrijving
`every( string $interval )`	Stel een intervalexpressie in (e.g, `1 hour`, `30 minutes`)
`cron( string $expression )`	Stel een cron-expressie met 5 velden in
`overlap( OverlapPolicy $policy )`	Stel het overlapbeleid in (standaard: `Allow`). Zie Planning.
`on_queue( string $queue )`	Stel de doelwachtrij in (standaard: `default`)
`id()`	Forceer het starten en geef de plannings-ID terug

`Queuety::scheduler()`

Haal de interne Scheduler-instantie op.

PHP

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

Handlerregistratie

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

Registreer een handlerklasse onder een naam.

PHP

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

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

Ontdek automatisch handlerklassen in een map en registreer ze. Geeft het aantal geregistreerde handlers terug.

PHP

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

`Queuety::registry()`

Haal het handlerregister op.

PHP

$registry = Queuety::registry();

Waarneembaarheid

`Queuety::logger()`

Haal de Logger-instantie op om logboekvermeldingen op te vragen.

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

Haal de Metrics-instantie op.

PHP

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

`Queuety::webhook_notifier()`

Haal de WebhookNotifier-instantie op.

PHP

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

`Queuety::rate_limiter()`

Haal de RateLimiter-instantie op.

PHP

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

Cache

`Queuety::cache()`

Haal de actieve instantie van de cachebackend op.

PHP

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

`Queuety::set_cache( $cache )`

Overschrijf de cachebackend. Roep dit aan vóór Queuety::init() om een aangepaste implementatie te gebruiken.

PHP

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

`QUEUETY_CACHE_TTL`

Standaard-TTL van de cache in seconden (standaard: 5). Stel deze constante in wp-config.php in om de standaard-TTL te wijzigen die interne cachebewerkingen gebruiken.

PHP

define( 'QUEUETY_CACHE_TTL', 10 );

Zie Caching voor meer informatie over backends en automatische detectie.

Gebeurtenislogboek van statusmachines

`Queuety::machine_events()`

Haal de StateMachineEventLog-instantie op.

PHP

$log = Queuety::machine_events();

Workflowgebeurtenislogboek

`Queuety::workflow_events()`

Haal de WorkflowEventLog-instantie op.

PHP

$log = Queuety::workflow_events();

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

Haal de volledige tijdlijn met gebeurtenissen voor een workflow op. Geeft een array met gebeurtenisrijen terug, gesorteerd op ID.

PHP

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

Elke gebeurtenisrij bevat: 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 (gekopieerd uit error.message), duration_ms en created_at.

`Queuety::workflow_trace( $workflow_id )`

Haal een genormaliseerde tracebundel op voor een workflowdebugger UI.

PHP

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

De trace bevat workflow, gegroepeerde steps, ruwe events, jobs, logs, artifacts, chunks, signals en wait_dependencies.

`Queuety::trace_input( $input )`

Stel input op applicatieniveau in voor de huidige uitvoeringsstap van de workflow.

`Queuety::trace_output( $output )`

Stel output op applicatieniveau in voor de huidige uitvoeringsstap van de workflow.

`Queuety::trace_context( array $context )`

Voeg context op applicatieniveau samen met de huidige uitvoeringsstap van de workflow.

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

Haal de openbare status op nadat een specifieke workflowstap is voltooid of hervat, of null als deze niet is gevonden.

PHP

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

Zie Workflowgebeurtenissenlog voor gebruiksscenario's en CLI-opdrachten.

Interne instanties

Deze methoden retourneren interne instanties voor geavanceerde gebruiksscenario's.

Methode	Retourneert	Beschrijving
`Queuety::queue()`	`Queue`	Wachtrijbewerkingen (claimen, vrijgeven, begraven)
`Queuety::worker()`	`Worker`	Beheer van workerprocessen
`Queuety::workflow_manager()`	`Workflow`	Orkestratie van workflows
`Queuety::machines()`	`StateMachine`	Beheer van de levenscyclus van de statusmachine
`Queuety::connection()`	`Connection`	Databaseverbinding
`Queuety::batch_manager()`	`BatchManager`	Beheer van de batchlevenscyclus (annuleren, opschonen)
`Queuety::chunk_store()`	`ChunkStore`	Persistentie van chunks voor streamingstappen
`Queuety::machine_events()`	`StateMachineEventLog`	Tracegebeurtenissen van statusmachines en genormaliseerde tracebundels
`Queuety::workflow_events()`	`WorkflowEventLog`	Tracegebeurtenissen van workflows en genormaliseerde tracebundels
`Queuety::resource_manager()`	`ResourceManager`	Toelatingsprofielen van workers, gewogen budgetten en inspectie van systeemgeheugen
`Queuety::cache()`	`Cache`	Cachebackendinstantie

Testen

`Queuety::fake()`

Vervang de wachtrij door een in-memory fake om te testen. Retourneert een instantie van QueueFake die verzonden taken registreert zonder de database aan te raken. create_batch() registreert ook fakiemetadata van batches; chains worden gecontroleerd aan de hand van de afzonderlijke gepushte taken die ze verzenden.

PHP

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

`Queuety::reset()`

Reset alle interne status, inclusief de wachtrij-fake. Roep aan in tearDown().

PHP

Queuety::reset();

Methoden van `QueueFake`

Methode	Beschrijving
`assert_pushed( string $class, ?Closure $callback )`	Controleer of een taakklasse is gepusht. Een optionele callback filtert op taakgegevens.
`assert_pushed_times( string $class, int $count )`	Controleer of een taak precies N keer is gepusht.
`assert_not_pushed( string $class )`	Controleer of een taak nooit is gepusht.
`assert_nothing_pushed()`	Controleer of er helemaal geen taken zijn gepusht.
`assert_batched( ?Closure $callback )`	Controleer of een batch is verzonden. Een optionele callback filtert op batchgegevens.
`pushed( string $class )`	Haal alle geregistreerde verzendingen voor een klasse op. Retourneert een array.
`batches()`	Haal alle geregistreerde batches op. Retourneert een array.
`reset()`	Wis alle geregistreerde taken en batches.

Zie Testen voor volledige voorbeelden.

Interfaces

`Handler`

Interface voor eenvoudige taakhandlers.

PHP

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

config() mag queue, max_attempts, backoff, rate_limit, concurrency_group, concurrency_limit en cost_units retourneren.

`Step`

Interface voor handlers van workflowstappen.

PHP

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

config() mag max_attempts, backoff, concurrency_group, concurrency_limit en cost_units retourneren.

`Contracts\ForEachHandler`

Interface voor dynamische for-each-branchhandlers.

PHP

namespace Queuety\Contracts;

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

config() mag max_attempts, backoff, rate_limit, concurrency_group, concurrency_limit en cost_units retourneren.

`Contracts\ForEachReducer`

Optionele reducer voor for-each-resultaten.

PHP

namespace Queuety\Contracts;

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

`Contracts\Compensation`

Interface voor handlers voor stapcompensatie.

PHP

namespace Queuety\Contracts;

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

`Contracts\RepeatCondition`

Interface voor loopvoorwaarden die worden gebruikt door repeat_until() en repeat_while() wanneer een vergelijking van statussleutels te beperkt is.

PHP

namespace Queuety\Contracts;

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

`Contracts\StateAction`

Interface voor acties bij het binnengaan van een status in de wachtrij.

PHP

namespace Queuety\Contracts;

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

$payload bevat de gestructureerde payload uit de actiedefinitie, als de machine daarmee is gedefinieerd. Retourneer updates van de openbare status. Voeg _event en optioneel _event_payload toe als de actie direct moet overgaan.

Statusacties kunnen ook een optionele methode config() definiëren. Queuety leest concurrency_group, concurrency_limit en cost_units daaruit wanneer de actie bij het binnengaan in de wachtrij wordt geplaatst.

`Contracts\StateGuard`

Interface voor bewakers van statusmachineovergangen.

PHP

namespace Queuety\Contracts;

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

`Contracts\Job`

Interface voor zelfstandige verzendbare taakklassen. Taakklassen kapselen hun payload in als openbare eigenschappen en hun uitvoeringslogica in handle().

PHP

namespace Queuety\Contracts;

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

Taakklassen kunnen optioneel een methode middleware() definiëren die een array met instanties van Contracts\Middleware retourneert.

`Contracts\Cache`

Interface voor cachebackends. Zie Caching voor ingebouwde implementaties en aangepaste backends.

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

Methode	Retourneert	Beschrijving
`get( $key )`	`mixed`	Haal een waarde op. Retourneert `null` als deze niet is gevonden of is verlopen.
`set( $key, $value, $ttl )`	`void`	Sla een waarde op. Een TTL van 0 betekent dat deze niet verloopt.
`delete( $key )`	`void`	Verwijder een waarde.
`has( $key )`	`bool`	Controleer of een sleutel bestaat en niet is verlopen.
`add( $key, $value, $ttl )`	`bool`	Atomair instellen als deze niet bestaat. Retourneert `true` als de waarde is ingesteld.
`flush()`	`void`	Verwijder alle items uit de cache.

`WorkflowEventLog`

Registreert overgangen tussen workflowstappen met de volledige tracecontext. Toegang via Queuety::workflow_events().

Methode	Retourneert	Beschrijving
`record_event( array $event )`	`void`	Registreer een ruwe tracegebeurtenis
`record_step_started( ... )`	`void`	Registreer een step_started-gebeurtenis met input en status vóór de uitvoering
`record_step_completed( ... )`	`void`	Registreer een step_completed-gebeurtenis met input, output, status vóór en status na de uitvoering
`record_step_failed( ... )`	`void`	Registreer een step_failed-gebeurtenis met tracecontext en gestructureerde foutdetails
`get_timeline( int $workflow_id, ?int $limit = 100, int $offset = 0 )`	`array`	Haal workflowtracegebeurtenissen in volgorde op, met optionele paginering
`get_trace( int $workflow_id )`	`array`	Haal een genormaliseerde tracebundel op
`get_state_at_step( int $workflow_id, int $step_index )`	`array\|null`	Haal de openbare status op nadat een specifieke stap is voltooid of hervat
`prune( int $older_than_days )`	`int`	Verwijder gebeurtenissen die ouder zijn dan het opgegeven aantal dagen. Retourneert het aantal verwijderde rijen.

`StateMachineEventLog`

Registreert statusmachinelevenscyclus-, actie-, overgangs- en bewakingsgebeurtenissen met dezelfde tracevelden als workflowgebeurtenissen.

Methode	Retourneert	Beschrijving
`record_event( array $event )`	`void`	Registreer een ruwe statusmachinetracegebeurtenis
`get_timeline( int $machine_id, ?int $limit = 100, int $offset = 0 )`	`array`	Haal statusmachinetracegebeurtenissen in volgorde op, met optionele paginering
`get_trace( int $machine_id )`	`array`	Haal een genormaliseerde statusmachinetracebundel op
`get_state_at_event( int $machine_id, int $event_id )`	`array\|null`	Haal de openbare status op na een specifieke statusmachinetracegebeurtenis
`prune( int $older_than_days )`	`int`	Verwijder gebeurtenissen die ouder zijn dan het opgegeven aantal dagen. Retourneert het aantal verwijderde rijen.

`Contracts\StreamingStep`

Interface voor handlers van workflowstappen die een stream van chunks produceren. Elke gegenereerde waarde wordt onmiddellijk in de database opgeslagen. Zie Streamingstappen.

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

Methode	Beschrijving
`stream( $state, $existing_chunks )`	Generator die stringchunks oplevert. Elke yield leidt meteen tot een databasewrite. Bij een nieuwe poging bevat `$existing_chunks` eerder opgeslagen chunks.
`on_complete( $chunks, $state )`	Wordt aangeroepen nadat de stream is voltooid. Retourneert gegevens om samen te voegen met de workflowstatus.
`config()`	Optionele configuratie. Ondersteunt `max_attempts`, `backoff`, `concurrency_group`, `concurrency_limit` en `cost_units`.

`ChunkStore`

Beheert de persistentie van chunks voor streamingstappen. Toegang via Queuety::chunk_store().

Methode	Retourneert	Beschrijving
`get_chunks( int $job_id )`	`string[]`	Haal alle chunks voor een taak op, geordend op chunkindex
`append_chunk( int $job_id, int $chunk_index, string $content, ?int $workflow_id, ?int $step_index )`	`void`	Voeg één chunk toe
`clear_chunks( int $job_id )`	`void`	Verwijder alle chunks voor een taak
`chunk_count( int $job_id )`	`int`	Tel de chunks voor een taak
`get_accumulated( int $job_id )`	`string`	Voeg alle chunks samen tot één string

`Heartbeat`

Statische helper voor het verzenden van activiteitenheartbeats vanuit handlers van stappen en streamingstappen. Zie Heartbeats.

Methode	Beschrijving
`Heartbeat::init( int $job_id, Connection $conn )`	Stel de taakcontext in. Wordt automatisch aangeroepen door de worker.
`Heartbeat::beat( array $progress = [] )`	Verzend een heartbeat. Werkt `reserved_at` bij en slaat optioneel voortgangsgegevens op in `heartbeat_data`.
`Heartbeat::clear()`	Wis de taakcontext. Wordt automatisch aangeroepen door de worker.
`Heartbeat::current_job_id()`	Haal de huidige taak-ID op (voor testen).

`Contracts\Middleware`

Interface voor taakmiddleware. Middleware omhult de taakuitvoering in een onion-achtige pipeline.

PHP

namespace Queuety\Contracts;

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

Traits

`Dispatchable`

Biedt statische verzendmethoden op taakklassen die Contracts\Job implementeren. Constructorargumenten worden doorgegeven om de taakinstantie te maken.

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

Methode	Retourneert	Beschrijving
`dispatch( ...$args )`	`PendingJob`	Maak een instantie en verzend deze
`dispatch_if( bool $condition, ...$args )`	`PendingJob\|null`	Verzend alleen als `$condition` waar is
`dispatch_unless( bool $condition, ...$args )`	`PendingJob\|null`	Verzend alleen als `$condition` onwaar is
`with_chain( array $jobs )`	`ChainBuilder`	Start een chain vanuit deze taakklasse

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

Taakeigenschappen

Taakklassen kunnen openbare eigenschappen declareren die de worker via reflectie uitleest om het gedrag voor nieuwe pogingen en time-outs aan te passen.

Eigenschap	Type	Beschrijving
`$tries`	`int`	Maximumaantal pogingen voordat de taak wordt begraven
`$timeout`	`int`	Maximale uitvoertijd in seconden (vereist `pcntl`)
`$backoff`	`array`	Oplopende vertragingen in seconden voor elke nieuwe poging

PHP

class ImportProductsJob implements Job {
    use Dispatchable;

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

    // ...
}

Hook `failed()`

Wanneer een taak wordt begraven nadat alle nieuwe pogingen zijn verbruikt, roept de worker een methode failed() aan op de taakinstantie, als die is gedefinieerd.

PHP

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

Ingebouwde middleware

Alle middlewareklassen implementeren Contracts\Middleware en bevinden zich in de namespace Queuety\Middleware.

Klasse	Beschrijving
`RateLimited( int $max, int $window )`	Dwing een limiet af van `$max` uitvoeringen per `$window` seconden
`Timeout( int $seconds )`	Beëindig de taak na `$seconds` seconden (vereist `pcntl`)
`UniqueJob( ?string $key )`	Voorkom gelijktijdige uitvoering via een databasevergrendeling
`WithoutOverlapping( string $key, int $release_after )`	Voorkom overlappende uitvoering met automatische vergrendelingsexpiratie
`ThrottlesExceptions( int $max_attempts, int $decay_minutes )`	Beperk uitzonderingen om taakstormen te voorkomen wanneer een externe service niet beschikbaar is

Zie Middleware voor gebruiksdetails en voorbeelden.

Enumeraties

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

Zie Planning voor meer informatie over de werking van elk beleid.

Attributen

`#[QueuetyHandler]`

PHP 8-attribuut voor automatische registratie. Zie PHP-attributen.

PHP

use Queuety\Attributes\QueuetyHandler;

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

CLI-bridge

`Queuety::cli_command()`

Retourneert de gepubliceerde WP-CLI-rootopdrachtnaam.

De standaard-rootopdracht is queuety. Definieer QUEUETY_CLI_COMMAND voordat Queuety start als de opdracht onder een andere naam moet worden gepubliceerd.

`Queuety::cli_command_map()`

Retourneert de serialiseerbare catalogus met Queuety CLI voor harnesses die het aangeboden opdrachtoppervlak moeten begrijpen zonder de klassen van WP-CLI uit te lezen.

Elke vermelding bevat:

een stabiele bewerkings-ID
het relatieve CLI-pad
de handlerklasse en -methode van WP-CLI
de mogelijke uitvoeringsdoelen
de adapter die ruwe argumenten van CLI normaliseert tot een uitvoeringsplan

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

Zet een geparseerde aanroep van CLI om in het uitvoeringsplan dat de harness moet uitvoeren.

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

Uitvoeringsplannen houden bewust rekening met het transport:

opdrachten worden omgezet in openbare aanroepbare functies van PHP met genormaliseerde argumenten
de catalogus vermeldt vooraf elk mogelijk uitvoeringsdoel, terwijl callable en transport de gekozen vertakking voor de geparseerde input aangeven

Dit is de standaardvorm die Queuety gebruikt voor agent-harnesses: een stabiele bewerkings-ID, een openbaar doel van API en normalisatie door de adapter voor zaken die alleen CLI betreffen, zoals standaardinstellingen voor wachtrijen, decodering van JSON-payloads, vertakking van workerpools en verwerking van workflow-exportbestanden.

PHP API

Toegang tot de zelfstandige kern

Queuety::init( Connection $conn )

Queuety::ensure_schema()

Jobs

Queuety::dispatch( $handler, $payload )

Methoden van PendingJob

Queuety::dispatch_sync( $handler, $payload )

Queuety::batch( $jobs )

Queuety::create_batch( $jobs )

Methoden van BatchBuilder

Queuety::find_batch( $id )

Workerbeheer

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

Waardeobject Batch

Queuety::chain( $jobs )

Methoden van ChainBuilder

Queuety::retry( $job_id )

Queuety::retry_buried()

Queuety::purge( $older_than_days )

Queues

Queuety::stats( $queue )

Queuety::buried( $queue )

Queuety::pause( $queue )

Queuety::resume( $queue )

Queuety::is_paused( $queue )

Workflows

Queuety::workflow( $name )

WorkflowBuilder-methoden

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 )