PHP API

Visualizza come Markdown Modifica su GitHub Documentazione di 2cc7b2b

Tutto in Queuety è accessibile tramite la classe statica Queuety\Queuety. È l'unico punto di accesso per avviare job, creare workflow, gestire macchine a stati e code e accedere ai sottosistemi interni.

PHP

use Queuety\Queuety;

Accesso autonomo al core

Il runtime basato su DB di Queuety può essere inizializzato al di fuori di WordPress.

Questo percorso autonomo è destinato al core di orchestrazione:

job e worker (i pool di worker richiedono anche le costanti DB_HOST, DB_NAME, DB_USER e DB_PASSWORD; vedi sotto)
workflow e segnali
macchine a stati ed eventi
artefatti, log, metriche e pianificazioni

Non sostituisce l'adattatore WordPress. Queuety::on_action(), replace_wp_cron(), l'inizializzazione del plugin e il percorso effettivo di registrazione wp queuety ... richiedono ancora WordPress.

`Queuety::init( Connection $conn )`

Inizializza Queuety con una connessione diretta basata 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 );

L'argomento facoltativo table_prefix modifica la parte Queuety dei nomi delle tabelle runtime per connessione:

PHP

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

Questo risolve tabelle come queuety_jobs in wp_themequeue_jobs.

`Queuety::ensure_schema()`

Verifica che le tabelle runtime di Queuety esistano per la connessione attiva.

PHP

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

Per il modello completo e il confine WordPress, consulta Uso autonomo.

Job

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

Avvia un job. Accetta una stringa con il nome di un handler oppure un'istanza Contracts\Job. Restituisce un builder PendingJob per concatenare le opzioni.

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

Quando viene passata un'istanza Contracts\Job, il serializzatore estrae il FQCN come nome dell'handler e le proprietà pubbliche come payload. Il parametro $payload viene ignorato quando $handler è un'istanza Job.

Metodi `PendingJob`

Metodo	Descrizione
`on_queue( string $queue )`	Imposta la coda di destinazione
`with_priority( Priority $priority )`	Imposta il livello di priorità
`delay( int $seconds )`	Ritardo prima che il job diventi disponibile
`max_attempts( int $max )`	Numero massimo di tentativi
`rate_limit( int $max, int $window )`	Limite di frequenza: numero massimo di esecuzioni per finestra di secondi
`unique()`	Impedisce avvii duplicati
`after( int $job_id )`	Attende prima il completamento di un altro job
`concurrency_group( string $group, int $limit )`	Inserisce il job in un gruppo condiviso di ammissione dei worker
`cost_units( int $units )`	Aggiunge metadati sul costo relativo di esecuzione all'avvio
`id()`	Forza l'avvio e restituisce l'ID del job

concurrency_group() e cost_units() alimentano il modello più ampio Gestione delle risorse, così lo stesso avvio può partecipare a limiti di concorrenza basati sul numero, budget ponderati per coda o provider e criteri euristici di ammissione dei worker.

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

Esegue un job in modo sincrono nel processo corrente, senza aggiungerlo alla coda.

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

Il metodo handle() del job viene chiamato direttamente. Non viene creata alcuna riga nel database, non interviene alcun worker e non si applica il middleware.

`Queuety::batch( $jobs )`

Avvia più job con un singolo INSERT a più righe.

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

Crea un BatchBuilder per avviare un gruppo di job con callback e monitoraggio dei progressi. Per tutti i dettagli, consulta Elaborazione in batch.

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

Metodi `BatchBuilder`

Metodo	Restituisce	Descrizione
`name( string $name )`	`self`	Imposta un nome per il batch
`on_queue( string $queue )`	`self`	Imposta la coda per tutti i job del batch
`then( string $handler_class )`	`self`	Handler chiamato al completamento riuscito del batch
`catch( string $handler_class )`	`self`	Handler chiamato quando il primo job del batch non riesce
`finally( string $handler_class )`	`self`	Handler chiamato al termine del batch, con successo o errore
`allow_failures()`	`self`	Consente l'esecuzione della callback `then` anche in caso di errori
`dispatch()`	`Batch`	Crea la riga del batch e tutti i job; restituisce l'oggetto valore `Batch`

`Queuety::find_batch( $id )`

Trova un batch tramite ID. Restituisce un oggetto valore Batch oppure null.

PHP

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

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

Controllo dei worker

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

Esegue direttamente un ciclo di worker.

PHP

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

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

Esegue direttamente un pool di worker di dimensione fissa. Richiede le estensioni pcntl e posix. Per le connessioni secondarie, il pool legge le costanti WordPress DB_HOST, DB_NAME, DB_USER e DB_PASSWORD (solo i prefissi delle tabelle provengono dalla connessione inizializzata); definiscile quando esegui un pool al di fuori di WordPress (#5).

PHP

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

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

Esegue direttamente un pool di worker adattivo. Ha gli stessi requisiti di run_worker_pool().

PHP

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

Oggetto valore `Batch`

Metodo/Proprietà	Tipo	Descrizione
`$id`	`int`	ID del batch
`$name`	`string\|null`	Nome facoltativo del batch
`$total_jobs`	`int`	Numero totale di job nel batch
`$pending_jobs`	`int`	Job ancora in attesa
`$failed_jobs`	`int`	Numero di job non riusciti
`$failed_job_ids`	`array`	ID dei job non riusciti
`$created_at`	`DateTimeImmutable`	Data di creazione del batch
`$finished_at`	`DateTimeImmutable\|null`	Data di completamento del batch
`$cancelled_at`	`DateTimeImmutable\|null`	Data di annullamento del batch
`progress()`	`int`	Percentuale di completamento (0-100)
`finished()`	`bool`	Indica se il batch è terminato
`has_failures()`	`bool`	Indica se un job non è riuscito
`cancelled()`	`bool`	Indica se il batch è stato annullato

`Queuety::chain( $jobs )`

Crea un ChainBuilder per l'esecuzione sequenziale dei job. Ogni job dipende dal completamento riuscito del precedente.

PHP

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

Metodi `ChainBuilder`

Metodo	Restituisce	Descrizione
`on_queue( string $queue )`	`self`	Imposta la coda per tutti i job della catena
`catch( string $handler_class )`	`self`	Handler chiamato quando un job della catena viene messo definitivamente da parte
`dispatch()`	`int`	Crea tutti i job con dipendenze sequenziali e restituisce l'ID del primo job

`Queuety::retry( $job_id )`

Riprova un job specifico. Lo riporta allo stato pending.

PHP

Queuety::retry( 42 );

`Queuety::retry_buried()`

Riprova tutti i job messi da parte. Restituisce il numero di job ritentati.

PHP

$count = Queuety::retry_buried();

`Queuety::purge( $older_than_days )`

Elimina i job completati. Restituisce il numero di job eliminati.

PHP

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

Code

`Queuety::stats( $queue )`

Ottieni il conteggio dei job raggruppati per stato.

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

Ottieni tutti i job messi da parte.

PHP

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

`Queuety::pause( $queue )`

Metti in pausa una coda, così i worker la saltano.

PHP

Queuety::pause( 'emails' );

`Queuety::resume( $queue )`

Riprendi una coda in pausa.

PHP

Queuety::resume( 'emails' );

`Queuety::is_paused( $queue )`

Verifica se una coda è in pausa.

PHP

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

Flussi di lavoro

`Queuety::workflow( $name )`

Inizia a creare un flusso di lavoro. Restituisce un WorkflowBuilder.

PHP

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

Metodi di `WorkflowBuilder`

Metodo	Descrizione
`then( string $handler_class, ?string $name = null )`	Aggiunge un passaggio sequenziale. Il nome è facoltativo e consente la diramazione condizionale.
`parallel( array $handler_classes )`	Aggiunge un gruppo di passaggi paralleli.
`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 )`	Aggiunge un passaggio dinamico «for each», che espande un ramo di lavoro per ogni elemento nello stato del flusso di lavoro.
`repeat_while( string $target_step, ?string $state_key = null, mixed $expected = true, ?string $name = null, ?string $condition_class = null, ?int $max_iterations = null )`	Torna a un passaggio precedente con nome finché una chiave dello stato pubblico o la classe `RepeatCondition` non corrisponde.
`repeat_until( string $target_step, ?string $state_key = null, mixed $expected = true, ?string $name = null, ?string $condition_class = null, ?int $max_iterations = null )`	Torna a un passaggio precedente con nome finché una chiave dello stato pubblico o la classe `RepeatCondition` non corrisponde.
`run_workflow( string $workflow_name, WorkflowBuilder $sub_builder )`	Aggiunge un passaggio per l'esecuzione di un flusso di lavoro.
`delay( int $seconds = 0, int $minutes = 0, int $hours = 0, int $days = 0 )`	Aggiunge un passaggio di attesa persistente. Le durate vengono sommate.
`wait_for_signal( string $name, ?string $result_key = null, ?string $step_name = null, array $match_payload = [], ?string $correlation_key = null )`	Aggiunge un passaggio di attesa di un segnale. `match_payload` e `correlation_key` sono facoltativi e consentono di ignorare gli eventi non correlati.
`wait_for_signals( array $signal_names, WaitMode $mode = WaitMode::All, ?string $result_key = null, ?string $name = null, array $match_payload = [], ?string $correlation_key = null )`	Attende tutti o alcuni tra più segnali, con corrispondenza facoltativa del payload e correlazione.
`wait_for_approval( string $signal_name = 'approval', ?string $result_key = 'approval', ?string $name = null, array $match_payload = [], ?string $correlation_key = null )`	Attende un segnale di approvazione umana.
`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 )`	Attende un'approvazione o un rifiuto e memorizza un payload decisionale strutturato.
`wait_for_input( string $signal_name = 'input', ?string $result_key = 'input', ?string $name = null, array $match_payload = [], ?string $correlation_key = null )`	Attende un segnale di input umano.
`wait_for_workflow( int\|string $workflow, ?string $result_key = null, ?string $name = null )`	Attende il completamento di un ID di flusso di lavoro o di un ID memorizzato nello stato pubblico.
`wait_for_workflows( array\|string $workflows, WaitMode $mode = WaitMode::All, ?string $result_key = null, ?string $name = null, ?int $quorum = null )`	Attende il completamento di tutti, alcuni o un quorum di più flussi di lavoro.
`wait_for_workflow_group( string $group_key, WaitMode $mode = WaitMode::All, ?int $quorum = null, ?string $result_key = null, ?string $name = null )`	Attende un gruppo di flussi di lavoro avviato in precedenza, usando la chiave interna del gruppo.
`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 )`	Avvia un flusso di lavoro di primo livello per ogni elemento di runtime; facoltativamente, contrassegna il gruppo avviato con una chiave persistente.
`start_agents( string $items_key, WorkflowBuilder $workflow_builder, string $result_key = 'agent_workflow_ids', string $payload_key = 'agent_task', bool $inherit_state = true, ?string $name = null, ?string $group_key = null )`	Alias orientato agli agenti per `start_workflows()`.
`wait_for_agents( array\|string $workflows = 'agent_workflow_ids', WaitMode $mode = WaitMode::All, ?string $result_key = 'agent_results', ?string $name = null, ?int $quorum = null )`	Alias orientato agli agenti per `wait_for_workflows()`.
`wait_for_agent_group( string $group_key = 'agents', WaitMode $mode = WaitMode::All, ?int $quorum = null, ?string $result_key = 'agent_results', ?string $name = null )`	Alias orientato agli agenti per `wait_for_workflow_group()`.
`version( string $version )`	Contrassegna la definizione del flusso di lavoro con una versione a livello di applicazione, utile per ispezioni ed esportazioni.
`idempotency_key( string $key )`	Rende persistente l'avvio del flusso di lavoro usando una chiave fornita dal chiamante. Un nuovo avvio con la stessa chiave restituisce l'ID del flusso di lavoro originale.
`max_transitions( int $max )`	Interrompe il flusso di lavoro se vengono completate più di `$max` transizioni di passaggio.
`max_for_each_items( int $max )`	Interrompe il flusso di lavoro se un passaggio `for_each()` tenta di espandere più di `$max` elementi.
`max_state_bytes( int $bytes )`	Interrompe il flusso di lavoro se lo stato pubblico supera `$bytes`.
`max_cost_units( int $max )`	Interrompe il flusso di lavoro se i passaggi completati superano `$max` unità di costo.
`max_started_workflows( int $max )`	Interrompe il flusso di lavoro prima di avviare più di `$max` flussi di lavoro secondari.
`compensate_with( string $handler_class )`	Collega un gestore di compensazione al passaggio aggiunto più di recente.
`compensate_on_failure()`	Esegue automaticamente le compensazioni dei passaggi completati in caso di errore del flusso di lavoro.
`on_queue( string $queue )`	Imposta la coda per tutti i passaggi.
`with_priority( Priority $priority )`	Imposta la priorità per tutti i passaggi.
`max_attempts( int $max )`	Imposta il numero massimo di tentativi per passaggio.
`on_cancel( string $handler_class )`	Registra una classe gestore di pulizia da usare in caso di annullamento. Vedi Annullamento.
`prune_state_after( int $steps = 2 )`	Abilita la rimozione automatica dei dati obsoleti dallo stato. Rimuove gli output dei passaggi più vecchi della soglia. Vedi Rimozione dei dati obsoleti dallo stato.
`must_complete_within( int $seconds = 0, int $minutes = 0, int $hours = 0, int $days = 0 )`	Imposta una scadenza per il flusso di lavoro. Le durate vengono sommate. Vedi Scadenze.
`on_deadline( string $handler_class )`	Registra un gestore chiamato quando la scadenza del flusso di lavoro viene superata. Vedi Scadenze.
`dispatch( array $initial_payload = [] )`	Avvia il flusso di lavoro. Restituisce l'ID del flusso di lavoro.

`Queuety::workflow_status( $workflow_id )`

Ottieni lo stato del flusso di lavoro. Restituisce un oggetto WorkflowState oppure 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 espone anche:

waiting_for: i nomi dei segnali o gli ID dei flussi di lavoro che al momento bloccano l'esecuzione
current_step_name: il nome configurato per il passaggio corrente, se presente
wait_mode: la modalità di attesa corrente (all, any o quorum) durante il blocco
wait_details: dati di ispezione aggiuntivi, ad esempio gli elementi bloccanti corrispondenti e quelli rimanenti
definition_version: il valore impostato da version(), se presente
definition_hash: un hash deterministico della definizione del flusso di lavoro
idempotency_key: la chiave persistente di avvio, se presente
budget: i limiti configurati e l'utilizzo attuale, se le protezioni sono abilitate
artifact_count: il numero di artefatti memorizzati per il flusso di lavoro
artifact_keys: le chiavi degli artefatti memorizzati, per ispezionarli rapidamente

`Queuety::retry_workflow( $workflow_id )`

Riprova un flusso di lavoro non riuscito a partire dal passaggio che ha causato l'errore.

PHP

Queuety::retry_workflow( $workflow_id );

Genera RuntimeException se il flusso di lavoro è già stato sottoposto a compensazione dopo un errore.

`Queuety::pause_workflow( $workflow_id )`

Mette in pausa un flusso di lavoro in esecuzione. Le attività già accodate per il passaggio corrente possono comunque terminare, ma i passaggi successivi non vengono accodati finché non si riprende il flusso di lavoro.

PHP

Queuety::pause_workflow( $workflow_id );

`Queuety::resume_workflow( $workflow_id )`

Riprende un flusso di lavoro in pausa.

PHP

Queuety::resume_workflow( $workflow_id );

`Queuety::cancel_workflow( $workflow_id )`

Annulla un flusso di lavoro ed esegue gli eventuali gestori di pulizia registrati. Imposta lo stato su cancelled e seppellisce tutte le attività in sospeso o in elaborazione del flusso di lavoro.

PHP

Queuety::cancel_workflow( $workflow_id );

Genera RuntimeException se il flusso di lavoro è già completed o cancelled. Per i dettagli, vedi Annullamento.

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

Riporta un flusso di lavoro a un passaggio precedente e lo esegue di nuovo da lì. Lo stato viene ripristinato dal state_after del registro eventi per il passaggio indicato, quindi viene accodato il passaggio successivo.

PHP

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

Genera RuntimeException se non esiste uno stato registrato per il passaggio di destinazione. Richiede il Registro eventi del flusso di lavoro. Per i dettagli, vedi Viaggio nel tempo.

`Queuety::fork_workflow( $workflow_id )`

Crea una copia indipendente di un flusso di lavoro in esecuzione, partendo dal suo stato corrente. Restituisce il nuovo ID del flusso di lavoro.

PHP

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

Il flusso di lavoro duplicato ha un ID proprio e opera indipendentemente dall'originale. Per i dettagli, vedi Creazione di rami.

`Queuety::export_workflow( $workflow_id )`

Esporta la cronologia completa di esecuzione di un flusso di lavoro come array serializzabile in JSON. Include la riga del flusso di lavoro, tutte le attività, gli eventi del flusso di lavoro e le voci di registro.

PHP

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

Per informazioni sulla struttura JSON e sui casi d'uso, vedi Esportazione e riproduzione.

`Queuety::replay_workflow( $data )`

Riproduce un flusso di lavoro esportato nell'ambiente corrente. Crea un nuovo flusso di lavoro dai dati di esportazione e ne restituisce l'ID.

PHP

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

Per i dettagli, vedi Esportazione e riproduzione.

`Queuety::replace_wp_cron()`

Sostituisce il sistema WordPress cron integrato con il pianificatore di Queuety. Intercetta wp_schedule_event(), wp_unschedule_event() e wp_get_scheduled_event() e indirizza gli eventi ricorrenti al worker Queuety.

PHP

Queuety::replace_wp_cron();

Funziona solo se WordPress è caricato. Per i dettagli, vedi WP-Cron Sostituzione.

`Queuety::restore_wp_cron()`

Ripristina il comportamento predefinito di WordPress cron rimuovendo gli hook del bridge Queuety cron.

PHP

Queuety::restore_wp_cron();

Per le limitazioni, vedi WP-Cron Sostituzione.

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

Invia un segnale a un flusso di lavoro. Se il flusso è in attesa di quel segnale, riprende immediatamente; altrimenti, il segnale viene memorizzato e utilizzato quando il flusso raggiunge il passaggio wait_for_signal corrispondente.

PHP

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

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

Per i dettagli, vedi Segnali del flusso di lavoro.

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

Wrapper semantico di signal() per i flussi di approvazione.

PHP

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

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

Wrapper semantico di signal() per i flussi di rifiuto.

PHP

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

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

Wrapper semantico di signal() per l'input umano strutturato.

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

Memorizza un artefatto persistente per un flusso di lavoro.

PHP

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

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

Memorizza un artefatto per il passaggio del flusso di lavoro attualmente in esecuzione.

PHP

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

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

Recupera un artefatto memorizzato per un flusso di lavoro.

PHP

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

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

Elenca gli artefatti di un flusso di lavoro.

PHP

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

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

Elimina un artefatto di un flusso di lavoro.

PHP

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

Macchine a stati

`Queuety::machine( $name )`

Inizia a creare una macchina a stati. Restituisce un StateMachineBuilder.

PHP

use Queuety\Enums\StateMachineStatus;

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

Metodi di `StateMachineBuilder`

Metodo	Descrizione
`initial( string $state_name )`	Imposta esplicitamente lo stato iniziale.
`state( string $state_name, ?StateMachineStatus $terminal_status = null )`	Avvia la configurazione di uno stato.
`action( string\|array $action_class )`	Registra un'azione di ingresso accodata per lo stato corrente. Passa il nome di una classe o una definizione strutturata con `class`, `payload` e `config`; `payload` raggiunge `StateAction::handle()` come `$payload`.
`on( string $event, string $target, string\|array\|null $guard_class = null, ?string $name = null )`	Registra una transizione di evento dallo stato corrente. La guardia accetta lo stesso formato strutturato di `action()`.
`on_queue( string $queue )`	Imposta la coda usata per le azioni di stato accodate.
`with_priority( Priority $priority )`	Imposta la priorità delle azioni di stato accodate.
`max_attempts( int $max_attempts )`	Imposta il limite di tentativi per le azioni di stato accodate.
`version( string $version )`	Contrassegna la definizione della macchina con una versione a livello di applicazione.
`idempotency_key( string $key )`	Rende persistente l'avvio con una chiave fornita dal chiamante. Un nuovo avvio con la stessa chiave restituisce l'ID della macchina originale.
`dispatch( array $initial_state = [] )`	Avvia la macchina e restituisce il relativo ID.

Le classi delle azioni di ingresso in uno stato possono esporre anche un metodo facoltativo config() con le stesse chiavi dei criteri delle risorse supportate dai passaggi del flusso di lavoro: concurrency_group, concurrency_limit e cost_units.

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

Avvia un pacchetto serializzato di definizione di una macchina a stati. La maggior parte del codice applicativo dovrebbe preferire machine()->dispatch(). Questo metodo di livello inferiore permette agli adattatori di avviare una macchina senza dipendere direttamente dal builder.

PHP

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

`Queuety::machine_status( $machine_id )`

Ottieni lo stato corrente della macchina. Restituisce un oggetto StateMachineState oppure 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 espone anche:

definition_version
definition_hash
idempotency_key
error_message
current_action
terminal_status

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

Invia un evento esterno a una macchina in attesa.

PHP

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

Il payload è disponibile per le guardie di transizione e, quando l'evento viene accettato, viene unito allo stato pubblico della macchina.

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

Elenca le righe persistite della macchina per consentirne l'ispezione.

PHP

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

Ogni riga include l'ID della macchina, il nome, lo stato, lo stato corrente, la versione della definizione, l'hash della definizione, la chiave di idempotenza e i timestamp del ciclo di vita.

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

Ottieni la cronologia persistente degli eventi di una macchina.

PHP

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

Ogni riga dell'evento contiene: 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 e created_at.

`Queuety::machine_trace( $machine_id )`

Ottieni un pacchetto di tracce normalizzato per un UI debugger di macchine a stati.

PHP

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

La traccia contiene machine, states raggruppati, events grezzi, jobs di azioni correlate, logs, artifacts e chunks.

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

Ottieni lo stato pubblico dopo uno specifico evento della traccia della macchina oppure null se non viene trovato.

PHP

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

Per informazioni sul modello del ciclo di vita e sui modelli d'uso, vedi Macchine a stati.

Modelli di flusso di lavoro

`Queuety::define_workflow( $name )`

Crea un builder di flusso di lavoro da usare come modello.

PHP

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

`Queuety::register_workflow_template( $builder )`

Registra un modello di flusso di lavoro.

PHP

Queuety::register_workflow_template( $builder );

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

Avvia un modello di flusso di lavoro registrato indicandone il nome. Restituisce l'ID del flusso di lavoro.

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

Genera RuntimeException se il modello non è registrato.

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

Avvia un flusso di lavoro da un pacchetto di definizione di runtime.

La maggior parte del codice applicativo dovrebbe preferire run_workflow() o workflow()->dispatch(). Questo metodo di livello inferiore permette agli adattatori riutilizzabili di avviare un flusso di lavoro da una definizione serializzata senza dipendere dai metodi interni del gestore dei flussi di lavoro.

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

Registra un'azione WordPress che avvia un flusso di lavoro.

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 può essere:

il nome di un modello di flusso di lavoro registrato
un WorkflowBuilder inline

map, when e idempotency_key vengono eseguiti in modo sincrono nell'ambito della richiesta corrente di WordPress. Solo i passaggi del flusso di lavoro vengono eseguiti in modo asincrono dopo l'avvio.

Per la guida completa, vedi Trigger di azione.

`Queuety::workflow_templates()`

Ottieni il registro dei modelli di flusso di lavoro.

PHP

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

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

Carica e registra tutti i file dei flussi di lavoro da una directory. Ogni file .php dovrebbe restituire un WorkflowBuilder. Restituisce il numero di flussi di lavoro registrati.

PHP

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

`Queuety::load_workflow_file( $file_path )`

Carica e registra un singolo file di flusso di lavoro. Restituisce il WorkflowTemplate oppure null se il file non restituisce un builder.

PHP

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

Pianificazione

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

Crea una pianificazione ricorrente. Restituisce un builder PendingSchedule.

PHP

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

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

Metodi di `PendingSchedule`

Metodo	Descrizione
`every( string $interval )`	Imposta un'espressione di intervallo (e.g. `1 hour`, `30 minutes`)
`cron( string $expression )`	Imposta un'espressione cron a 5 campi
`overlap( OverlapPolicy $policy )`	Imposta la politica di sovrapposizione (predefinita: `Allow`). Vedi Pianificazione.
`on_queue( string $queue )`	Imposta la coda di destinazione (predefinita: `default`)
`id()`	Forza l'avvio e restituisce l'ID della pianificazione

`Queuety::scheduler()`

Ottieni l'istanza interna di Scheduler.

PHP

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

Registrazione dei gestori

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

Registra una classe gestore con un nome.

PHP

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

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

Individua automaticamente e registra le classi gestore da una directory. Restituisce il numero di gestori registrati.

PHP

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

`Queuety::registry()`

Ottieni il registro dei gestori.

PHP

$registry = Queuety::registry();

Osservabilità

`Queuety::logger()`

Ottieni l'istanza Logger per interrogare le voci di registro.

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

Ottieni l'istanza Metrics.

PHP

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

`Queuety::webhook_notifier()`

Ottieni l'istanza WebhookNotifier.

PHP

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

`Queuety::rate_limiter()`

Ottieni l'istanza RateLimiter.

PHP

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

Cache

`Queuety::cache()`

Ottieni l'istanza del backend di cache attivo.

PHP

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

`Queuety::set_cache( $cache )`

Sostituisci il backend di cache. Chiama Queuety::init() prima di questo metodo per usare un'implementazione personalizzata.

PHP

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

`QUEUETY_CACHE_TTL`

TTL predefinito della cache in secondi (valore predefinito: 5). Imposta questa costante in wp-config.php per cambiare il TTL predefinito usato dalle operazioni interne della cache.

PHP

define( 'QUEUETY_CACHE_TTL', 10 );

Per informazioni sui backend e sul rilevamento automatico, vedi Memorizzazione nella cache.

Registro eventi della macchina a stati

`Queuety::machine_events()`

Ottieni l'istanza StateMachineEventLog.

PHP

$log = Queuety::machine_events();

Registro eventi del flusso di lavoro

`Queuety::workflow_events()`

Ottieni l'istanza WorkflowEventLog.

PHP

$log = Queuety::workflow_events();

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

Ottieni la cronologia completa degli eventi di un flusso di lavoro. Restituisce un array di righe degli eventi ordinate per ID.

PHP

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

Ogni riga dell'evento contiene: 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 (copiato da error.message), duration_ms e created_at.

`Queuety::workflow_trace( $workflow_id )`

Ottenete un pacchetto di tracce normalizzate per un debugger di workflow UI.

PHP

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

La traccia contiene workflow, raggruppati per steps, events grezzi, jobs, logs, artifacts, chunks, signals e wait_dependencies.

`Queuety::trace_input( $input )`

Imposta l'input a livello di applicazione per il passaggio del workflow in esecuzione.

`Queuety::trace_output( $output )`

Imposta l'output a livello di applicazione per il passaggio del workflow in esecuzione.

`Queuety::trace_context( array $context )`

Unisce il contesto a livello di applicazione al passaggio del workflow in esecuzione.

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

Ottiene lo stato pubblico dopo il completamento o la ripresa di uno specifico passaggio del workflow, oppure null se non viene trovato.

PHP

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

Consultate Registro degli eventi del workflow per i casi d'uso e i comandi CLI.

Istanze interne

Questi metodi restituiscono istanze interne per casi d'uso avanzati.

Metodo	Restituisce	Descrizione
`Queuety::queue()`	`Queue`	Operazioni sulla coda (presa in carico, rilascio, spostamento in archivio)
`Queuety::worker()`	`Worker`	Controllo del processo worker
`Queuety::workflow_manager()`	`Workflow`	Orchestrazione del workflow
`Queuety::machines()`	`StateMachine`	Gestione del ciclo di vita della macchina a stati
`Queuety::connection()`	`Connection`	Connessione al database
`Queuety::batch_manager()`	`BatchManager`	Gestione del ciclo di vita dei batch (annullamento, pulizia)
`Queuety::chunk_store()`	`ChunkStore`	Persistenza dei chunk per i passaggi in streaming
`Queuety::machine_events()`	`StateMachineEventLog`	Eventi di traccia della macchina a stati e pacchetti di tracce normalizzate
`Queuety::workflow_events()`	`WorkflowEventLog`	Eventi di traccia del workflow e pacchetti di tracce normalizzate
`Queuety::resource_manager()`	`ResourceManager`	Profili di ammissione dei worker, budget ponderati e ispezione della memoria di sistema
`Queuety::cache()`	`Cache`	Istanza del backend di cache

Test

`Queuety::fake()`

Sostituisce la coda con un'implementazione fittizia in memoria per i test. Restituisce un'istanza QueueFake che registra i job inviati senza interagire con il database. create_batch() registra anche metadati fittizi dei batch; le catene vengono verificate tramite i singoli job inviati.

PHP

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

`Queuety::reset()`

Reimposta tutto lo stato interno, inclusa la coda fittizia. Chiamare in tearDown().

PHP

Queuety::reset();

Metodi di `QueueFake`

Metodo	Descrizione
`assert_pushed( string $class, ?Closure $callback )`	Verifica che sia stata inviata una classe di job. Un callback facoltativo filtra i dati del job.
`assert_pushed_times( string $class, int $count )`	Verifica che un job sia stato inviato esattamente N volte.
`assert_not_pushed( string $class )`	Verifica che un job non sia mai stato inviato.
`assert_nothing_pushed()`	Verifica che non sia stato inviato alcun job.
`assert_batched( ?Closure $callback )`	Verifica che sia stato inviato un batch. Un callback facoltativo filtra i dati del batch.
`pushed( string $class )`	Ottiene tutti gli invii registrati per una classe. Restituisce un array.
`batches()`	Ottiene tutti i batch registrati. Restituisce un array.
`reset()`	Cancella tutti i job e i batch registrati.

Consultate Test per esempi completi.

Interfacce

`Handler`

Interfaccia per gestori di job semplici.

PHP

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

config() può restituire queue, max_attempts, backoff, rate_limit, concurrency_group, concurrency_limit e cost_units.

`Step`

Interfaccia per gestori dei passaggi del workflow.

PHP

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

config() può restituire max_attempts, backoff, concurrency_group, concurrency_limit e cost_units.

`Contracts\ForEachHandler`

Interfaccia per gestori di rami dinamici for-each.

PHP

namespace Queuety\Contracts;

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

config() può restituire max_attempts, backoff, rate_limit, concurrency_group, concurrency_limit e cost_units.

`Contracts\ForEachReducer`

Riduttore facoltativo per i risultati di for-each.

PHP

namespace Queuety\Contracts;

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

`Contracts\Compensation`

Interfaccia per i gestori di compensazione dei passaggi.

PHP

namespace Queuety\Contracts;

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

`Contracts\RepeatCondition`

Interfaccia per i predicati dei cicli usati da repeat_until() e repeat_while() quando un confronto tra chiavi di stato è troppo limitato.

PHP

namespace Queuety\Contracts;

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

`Contracts\StateAction`

Interfaccia per le azioni di ingresso nello stato messe in coda.

PHP

namespace Queuety\Contracts;

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

$payload riceve la struttura payload dalla definizione dell'azione, se la macchina è stata definita con una. Restituisce gli aggiornamenti dello stato pubblico. Includete _event e, facoltativamente, _event_payload se l'azione deve avviare subito una transizione.

Le azioni di stato possono anche definire un metodo facoltativo config(). Queuety ne legge concurrency_group, concurrency_limit e cost_units quando l'azione di ingresso viene messa in coda.

`Contracts\StateGuard`

Interfaccia per le condizioni di guardia delle transizioni della macchina a stati.

PHP

namespace Queuety\Contracts;

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

`Contracts\Job`

Interfaccia per classi di job inviabili autonome. Le classi di job racchiudono il proprio payload in proprietà pubbliche e la logica di esecuzione in handle().

PHP

namespace Queuety\Contracts;

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

Le classi di job possono definire facoltativamente un metodo middleware() che restituisce un array di istanze Contracts\Middleware.

`Contracts\Cache`

Interfaccia per i backend di cache. Consultate Caching per le implementazioni integrate e i backend personalizzati.

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

Metodo	Restituisce	Descrizione
`get( $key )`	`mixed`	Recupera un valore. Restituisce `null` se non viene trovato o è scaduto.
`set( $key, $value, $ttl )`	`void`	Memorizza un valore. Un TTL pari a 0 indica che non scade.
`delete( $key )`	`void`	Rimuove un valore.
`has( $key )`	`bool`	Verifica se una chiave esiste e non è scaduta.
`add( $key, $value, $ttl )`	`bool`	Imposta atomicamente solo se la chiave non esiste. Restituisce `true` se il valore è stato impostato.
`flush()`	`void`	Rimuove tutti gli elementi dalla cache.

`WorkflowEventLog`

Registra le transizioni dei passaggi del workflow con il contesto completo della traccia. Accessibile tramite Queuety::workflow_events().

Metodo	Restituisce	Descrizione
`record_event( array $event )`	`void`	Registra un evento di traccia grezzo
`record_step_started( ... )`	`void`	Registra un evento step_started con input e stato prima dell'esecuzione
`record_step_completed( ... )`	`void`	Registra un evento step_completed con input, output, stato prima e dopo l'esecuzione
`record_step_failed( ... )`	`void`	Registra un evento step_failed con il contesto della traccia e dettagli strutturati sull'errore
`get_timeline( int $workflow_id, ?int $limit = 100, int $offset = 0 )`	`array`	Ottiene gli eventi di traccia del workflow in ordine, con paginazione facoltativa
`get_trace( int $workflow_id )`	`array`	Ottiene un pacchetto di tracce normalizzate
`get_state_at_step( int $workflow_id, int $step_index )`	`array\|null`	Ottiene lo stato pubblico dopo il completamento o la ripresa di un passaggio specifico
`prune( int $older_than_days )`	`int`	Elimina gli eventi più vecchi del numero di giorni indicato. Restituisce il numero di righe eliminate.

`StateMachineEventLog`

Registra il ciclo di vita, le azioni, le transizioni e gli eventi delle condizioni di guardia della macchina a stati, usando gli stessi campi di traccia degli eventi del workflow.

Metodo	Restituisce	Descrizione
`record_event( array $event )`	`void`	Registra un evento di traccia grezzo della macchina
`get_timeline( int $machine_id, ?int $limit = 100, int $offset = 0 )`	`array`	Ottiene gli eventi di traccia della macchina in ordine, con paginazione facoltativa
`get_trace( int $machine_id )`	`array`	Ottiene un pacchetto di tracce normalizzate della macchina
`get_state_at_event( int $machine_id, int $event_id )`	`array\|null`	Ottiene lo stato pubblico dopo uno specifico evento di traccia della macchina
`prune( int $older_than_days )`	`int`	Elimina gli eventi più vecchi del numero di giorni indicato. Restituisce il numero di righe eliminate.

`Contracts\StreamingStep`

Interfaccia per gestori dei passaggi del workflow che producono uno stream di chunk. Ogni valore generato viene salvato subito nel database. Consultate Passaggi in streaming.

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

Metodo	Descrizione
`stream( $state, $existing_chunks )`	Generatore che produce chunk di stringhe. Ogni valore generato avvia una scrittura immediata nel DB. In caso di nuovo tentativo, `$existing_chunks` contiene i chunk salvati in precedenza.
`on_complete( $chunks, $state )`	Viene chiamato al termine dello stream. Restituisce dati da unire allo stato del workflow.
`config()`	Configurazione facoltativa. Supporta `max_attempts`, `backoff`, `concurrency_group`, `concurrency_limit` e `cost_units`.

`ChunkStore`

Gestisce la persistenza dei chunk per i passaggi in streaming. Accessibile tramite Queuety::chunk_store().

Metodo	Restituisce	Descrizione
`get_chunks( int $job_id )`	`string[]`	Recupera tutti i chunk di un job, ordinati per indice del chunk
`append_chunk( int $job_id, int $chunk_index, string $content, ?int $workflow_id, ?int $step_index )`	`void`	Aggiunge un singolo chunk
`clear_chunks( int $job_id )`	`void`	Elimina tutti i chunk di un job
`chunk_count( int $job_id )`	`int`	Conta i chunk di un job
`get_accumulated( int $job_id )`	`string`	Concatena tutti i chunk in un'unica stringa

`Heartbeat`

Helper statico per inviare heartbeat di attività dall'interno dei gestori dei passaggi e dei passaggi in streaming. Consultate Heartbeat.

Metodo	Descrizione
`Heartbeat::init( int $job_id, Connection $conn )`	Imposta il contesto del job. Viene chiamato automaticamente dal worker.
`Heartbeat::beat( array $progress = [] )`	Invia un heartbeat. Aggiorna `reserved_at` e memorizza facoltativamente i dati di avanzamento in `heartbeat_data`.
`Heartbeat::clear()`	Cancella il contesto del job. Viene chiamato automaticamente dal worker.
`Heartbeat::current_job_id()`	Ottiene l'ID del job corrente (per i test).

`Contracts\Middleware`

Interfaccia per il middleware dei job. Il middleware avvolge l'esecuzione dei job in una pipeline a cipolla.

PHP

namespace Queuety\Contracts;

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

Trait

`Dispatchable`

Fornisce metodi statici di invio nelle classi di job che implementano Contracts\Job. Gli argomenti del costruttore vengono passati per creare l'istanza del job.

PHP

use Queuety\Contracts\Job;
use Queuety\Dispatchable;

class SendEmailJob implements Job {
    use Dispatchable;
    // ...
}

// Dispatches a new SendEmailJob instance
SendEmailJob::dispatch( 'user@example.com', 'Subject', 'Body' );

Metodo	Restituisce	Descrizione
`dispatch( ...$args )`	`PendingJob`	Crea un'istanza e la invia
`dispatch_if( bool $condition, ...$args )`	`PendingJob\|null`	La invia solo se `$condition` è true
`dispatch_unless( bool $condition, ...$args )`	`PendingJob\|null`	La invia solo se `$condition` è false
`with_chain( array $jobs )`	`ChainBuilder`	Avvia una catena da questa classe di job

PHP

// Conditional dispatch
SendEmailJob::dispatch_if( $user->wants_email, $user->email, 'Hello', 'Hi!' );
SendEmailJob::dispatch_unless( $maintenance_mode, $user->email, 'Hello', 'Hi!' );

// Chain from a job class
FetchDataJob::with_chain( [
    new ProcessDataJob(),
    new NotifyCompleteJob(),
] )->dispatch();

Proprietà del job

Le classi di job possono dichiarare proprietà pubbliche che il worker legge tramite reflection per modificare il comportamento dei nuovi tentativi e dei timeout.

Proprietà	Tipo	Descrizione
`$tries`	`int`	Numero massimo di tentativi prima che il job venga spostato in archivio
`$timeout`	`int`	Tempo massimo di esecuzione in secondi (richiede `pcntl`)
`$backoff`	`array`	Intervalli di attesa crescenti in secondi per ogni tentativo successivo

PHP

class ImportProductsJob implements Job {
    use Dispatchable;

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

    // ...
}

Hook `failed()`

Quando un job viene spostato in archivio dopo aver esaurito tutti i tentativi, il worker chiama un metodo failed() sull'istanza del job, se definito.

PHP

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

Middleware integrato

Tutte le classi middleware implementano Contracts\Middleware e si trovano nel namespace Queuety\Middleware.

Classe	Descrizione
`RateLimited( int $max, int $window )`	Impone un limite di `$max` esecuzioni ogni `$window` secondi
`Timeout( int $seconds )`	Termina il job dopo `$seconds` secondi (richiede `pcntl`)
`UniqueJob( ?string $key )`	Impedisce l'esecuzione simultanea tramite un lock del database
`WithoutOverlapping( string $key, int $release_after )`	Impedisce le esecuzioni sovrapposte con la scadenza automatica del lock
`ThrottlesExceptions( int $max_attempts, int $decay_minutes )`	Limita le eccezioni per prevenire picchi di job quando un servizio esterno non è disponibile

Consultate Middleware per dettagli ed esempi d'uso.

Enum

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

Consultate Pianificazione per i dettagli sul funzionamento di ciascun criterio.

Attributi

`#[QueuetyHandler]`

PHP 8 attributo per la registrazione automatica. Consultate PHP Attributi.

PHP

use Queuety\Attributes\QueuetyHandler;

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

Bridge CLI

`Queuety::cli_command()`

Restituisce il nome del comando root WP-CLI pubblicato.

Il comando root predefinito è queuety. Definite QUEUETY_CLI_COMMAND prima dell'avvio di Queuety se il comando deve essere pubblicato con un nome diverso.

`Queuety::cli_command_map()`

Restituisce il catalogo serializzabile dei comandi Queuety CLI per gli harness che devono comprendere i comandi forniti senza analizzare le classi WP-CLI.

Ogni voce include:

un ID operazione stabile
il percorso relativo CLI
la classe e il metodo del gestore WP-CLI
i possibili obiettivi di esecuzione
l'adapter che normalizza gli argomenti grezzi CLI in un piano di esecuzione

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

Trasforma un'invocazione CLI analizzata nel piano di esecuzione che l'harness deve eseguire.

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

I piani di esecuzione tengono conto del trasporto:

i comandi si risolvono in callable pubblici PHP con argomenti normalizzati
il catalogo elenca in anticipo tutti i possibili obiettivi di esecuzione, mentre callable e transport identificano il ramo scelto per l'input analizzato

Questo è lo schema standard Queuety per gli harness degli agenti: ID operazione stabile, obiettivo pubblico API e normalizzazione gestita dall'adapter per gli aspetti specifici di CLI, come i valori predefiniti della coda, la decodifica del payload JSON, la selezione del ramo del worker pool e la gestione dei file di esportazione del workflow.

PHP API

Accesso autonomo al core

Queuety::init( Connection $conn )

Queuety::ensure_schema()

Job

Queuety::dispatch( $handler, $payload )

Metodi PendingJob

Queuety::dispatch_sync( $handler, $payload )

Queuety::batch( $jobs )

Queuety::create_batch( $jobs )

Metodi BatchBuilder

Queuety::find_batch( $id )

Controllo dei worker

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

Oggetto valore Batch

Queuety::chain( $jobs )

Metodi ChainBuilder

Queuety::retry( $job_id )

Queuety::retry_buried()

Queuety::purge( $older_than_days )

Code

Queuety::stats( $queue )

Queuety::buried( $queue )

Queuety::pause( $queue )

Queuety::resume( $queue )

Queuety::is_paused( $queue )

Flussi di lavoro

Queuety::workflow( $name )

Metodi di WorkflowBuilder

Queuety::workflow_status( $workflow_id )

Queuety::retry_workflow( $workflow_id )

Queuety::pause_workflow( $workflow_id )

Queuety::resume_workflow( $workflow_id )

Queuety::cancel_workflow( $workflow_id )

Queuety::rewind_workflow( $workflow_id, $to_step )

Queuety::fork_workflow( $workflow_id )

Queuety::export_workflow( $workflow_id )

Queuety::replay_workflow( $data )

Queuety::replace_wp_cron()

Queuety::restore_wp_cron()

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

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

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

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

Queuety::put_artifact( $workflow_id, $artifact_key, $content, $kind = 'json', $step_index = null, $metadata = [] )

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

Queuety::workflow_artifact( $workflow_id, $artifact_key )

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

Queuety::delete_workflow_artifact( $workflow_id, $artifact_key )