PHP API

Wyświetl jako Markdown Edytuj w GitHub Dokumentacja: 2cc7b2b

Wszystko w Queuety jest dostępne za pośrednictwem statycznej klasy Queuety\Queuety. To jedyny punkt wejścia do zlecania zadań, tworzenia przepływów pracy, sterowania maszynami stanów, zarządzania kolejkami i uzyskiwania dostępu do podsystemów wewnętrznych.

PHP

use Queuety\Queuety;

Dostęp do samodzielnego rdzenia

Środowisko uruchomieniowe Queuety oparte na bazie danych można zainicjować poza WordPress.

Ta samodzielna ścieżka jest przeznaczona dla rdzenia orkiestracji:

zadania i procesy robocze (pule procesów roboczych wymagają też stałych DB_HOST, DB_NAME, DB_USER i DB_PASSWORD; patrz niżej)
przepływy pracy i sygnały
maszyny stanów i zdarzenia
artefakty, logi, metryki i harmonogramy

To nie zastępuje adaptera WordPress. Queuety::on_action(), replace_wp_cron(), inicjalizacja wtyczki i właściwa ścieżka rejestracji wp queuety ... nadal wymagają WordPress.

`Queuety::init( Connection $conn )`

Zainicjuj Queuety z bezpośrednim połączeniem opartym na bazie danych.

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

Opcjonalny argument table_prefix zmienia część nazw tabel środowiska uruchomieniowego Queuety dla każdego połączenia:

PHP

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

Dzięki temu tabele takie jak queuety_jobs są rozwiązywane do wp_themequeue_jobs.

`Queuety::ensure_schema()`

Upewnij się, że tabele środowiska uruchomieniowego Queuety istnieją dla aktywnego połączenia.

PHP

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

Pełny model i granicę WordPress opisano w sekcji Samodzielne użycie.

Zadania

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

Zleć zadanie. Akceptuje ciąg znaków z nazwą procedury obsługi lub instancję Contracts\Job. Zwraca obiekt budujący PendingJob, umożliwiający łańcuchowe przekazywanie opcji.

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

Gdy przekazana zostanie instancja Contracts\Job, serializator pobiera FQCN jako nazwę procedury obsługi, a właściwości publiczne jako dane wejściowe. Parametr $payload jest ignorowany, gdy $handler jest instancją Job.

Metody `PendingJob`

Metoda	Opis
`on_queue( string $queue )`	Ustaw docelową kolejkę
`with_priority( Priority $priority )`	Ustaw poziom priorytetu
`delay( int $seconds )`	Opóźnienie przed udostępnieniem zadania
`max_attempts( int $max )`	Maksymalna liczba prób ponowienia
`rate_limit( int $max, int $window )`	Limit szybkości: maksymalna liczba wykonań w oknie liczonym w sekundach
`unique()`	Zapobiegaj wielokrotnemu zlecaniu tego samego zadania
`after( int $job_id )`	Poczekaj, aż inne zadanie zakończy się jako pierwsze
`concurrency_group( string $group, int $limit )`	Umieść zadanie we współdzielonej grupie dopuszczania procesów roboczych
`cost_units( int $units )`	Dołącz metadane względnego kosztu wykonania do zlecenia
`id()`	Wymuś zlecenie i zwróć ID zadania

concurrency_group() i cost_units() są częścią szerszego modelu Zarządzanie zasobami, dzięki czemu to samo zlecenie może podlegać limitom współbieżności opartym na liczbie zadań, ważonym budżetom kolejki lub dostawcy oraz heurystykom dopuszczania procesów roboczych.

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

Wykonaj zadanie synchronicznie w bieżącym procesie, bez dodawania go do kolejki.

PHP

// With a Job instance
Queuety::dispatch_sync( new SendEmailJob( to: 'user@example.com', subject: 'Hi', body: 'Hello' ) );

// With a handler name and payload
Queuety::dispatch_sync( 'send_email', [ 'to' => 'user@example.com' ] );

Metoda handle() zadania jest wywoływana bezpośrednio. Nie jest tworzony wiersz w bazie danych, nie uczestniczy w tym żaden proces roboczy, a oprogramowanie pośredniczące nie jest stosowane.

`Queuety::batch( $jobs )`

Zleć wiele zadań w jednym wielowierszowym poleceniu INSERT.

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

Utwórz obiekt BatchBuilder do zlecania grupy zadań z wywołaniami zwrotnymi i śledzeniem postępu. Szczegóły znajdziesz w sekcji Przetwarzanie partiami.

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

Metody `BatchBuilder`

Metoda	Zwraca	Opis
`name( string $name )`	`self`	Ustaw nazwę partii
`on_queue( string $queue )`	`self`	Ustaw kolejkę dla wszystkich zadań w partii
`then( string $handler_class )`	`self`	Procedura obsługi wywoływana po pomyślnym ukończeniu partii
`catch( string $handler_class )`	`self`	Procedura obsługi wywoływana po niepowodzeniu pierwszego zadania w partii
`finally( string $handler_class )`	`self`	Procedura obsługi wywoływana po zakończeniu partii (sukcesem lub niepowodzeniem)
`allow_failures()`	`self`	Zezwól na uruchomienie wywołania zwrotnego `then` mimo niepowodzeń
`dispatch()`	`Batch`	Utwórz wiersz partii i wszystkie zadania; zwraca obiekt wartości `Batch`

`Queuety::find_batch( $id )`

Znajdź partię po ID. Zwraca obiekt wartości Batch lub null.

PHP

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

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

Sterowanie procesami roboczymi

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

Uruchom bezpośrednio jedną pętlę procesu roboczego.

PHP

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

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

Uruchom bezpośrednio pulę procesów roboczych o stałym rozmiarze. Wymaga rozszerzeń pcntl i posix. Dla połączeń podrzędnych pula odczytuje stałe WordPress, DB_HOST, DB_NAME, DB_USER i DB_PASSWORD (tylko prefiksy tabel pochodzą z zainicjowanego połączenia), dlatego zdefiniuj je podczas uruchamiania puli poza WordPress (#5).

PHP

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

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

Uruchom bezpośrednio adaptacyjną pulę procesów roboczych. Obowiązują te same wymagania co w przypadku run_worker_pool().

PHP

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

Obiekt wartości `Batch`

Metoda/właściwość	Typ	Opis
`$id`	`int`	ID partii
`$name`	`string\|null`	Opcjonalna nazwa partii
`$total_jobs`	`int`	Łączna liczba zadań w partii
`$pending_jobs`	`int`	Zadania oczekujące na wykonanie
`$failed_jobs`	`int`	Liczba nieudanych zadań
`$failed_job_ids`	`array`	ID nieudanych zadań
`$created_at`	`DateTimeImmutable`	Data utworzenia partii
`$finished_at`	`DateTimeImmutable\|null`	Data zakończenia partii
`$cancelled_at`	`DateTimeImmutable\|null`	Data anulowania partii
`progress()`	`int`	Procent ukończenia (0-100)
`finished()`	`bool`	Czy partia została zakończona
`has_failures()`	`bool`	Czy jakiekolwiek zadanie zakończyło się niepowodzeniem
`cancelled()`	`bool`	Czy partia została anulowana

`Queuety::chain( $jobs )`

Utwórz obiekt ChainBuilder do sekwencyjnego wykonywania zadań. Każde zadanie zależy od pomyślnego ukończenia poprzedniego.

PHP

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

Metody `ChainBuilder`

Metoda	Zwraca	Opis
`on_queue( string $queue )`	`self`	Ustaw kolejkę dla wszystkich zadań w łańcuchu
`catch( string $handler_class )`	`self`	Procedura obsługi wywoływana po trwałym przeniesieniu dowolnego zadania w łańcuchu do stanu ukrytego
`dispatch()`	`int`	Utwórz wszystkie zadania z zależnościami sekwencyjnymi; zwraca ID pierwszego zadania

`Queuety::retry( $job_id )`

Ponów próbę wykonania określonego zadania. Przywraca mu status pending.

PHP

Queuety::retry( 42 );

`Queuety::retry_buried()`

Ponów próbę wykonania wszystkich ukrytych zadań. Zwraca liczbę zadań, których próbę ponowiono.

PHP

$count = Queuety::retry_buried();

`Queuety::purge( $older_than_days )`

Usuń ukończone zadania. Zwraca liczbę usuniętych zadań.

PHP

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

Kolejki

`Queuety::stats( $queue )`

Pobierz liczbę zadań pogrupowanych według statusu.

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

Pobierz wszystkie ukryte zadania.

PHP

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

`Queuety::pause( $queue )`

Wstrzymaj kolejkę, aby procesy robocze ją pomijały.

PHP

Queuety::pause( 'emails' );

`Queuety::resume( $queue )`

Wznów wstrzymaną kolejkę.

PHP

Queuety::resume( 'emails' );

`Queuety::is_paused( $queue )`

Sprawdź, czy kolejka jest wstrzymana.

PHP

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

Przepływy pracy

`Queuety::workflow( $name )`

Rozpocznij tworzenie przepływu pracy. Zwraca obiekt WorkflowBuilder.

PHP

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

Metody `WorkflowBuilder`

Metoda	Opis
`then( string $handler_class, ?string $name = null )`	Dodaj krok sekwencyjny. Opcjonalna nazwa służy do warunkowego rozgałęziania.
`parallel( array $handler_classes )`	Dodaj grupę kroków równoległych.
`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 )`	Dodaj dynamiczny krok iteracji, który tworzy odrębne zadanie gałęzi dla każdego elementu w stanie przepływu pracy.
`repeat_while( string $target_step, ?string $state_key = null, mixed $expected = true, ?string $name = null, ?string $condition_class = null, ?int $max_iterations = null )`	Wróć do wcześniejszego nazwanego kroku, dopóki publiczny klucz stanu lub klasa `RepeatCondition` nie będzie pasować.
`repeat_until( string $target_step, ?string $state_key = null, mixed $expected = true, ?string $name = null, ?string $condition_class = null, ?int $max_iterations = null )`	Wróć do wcześniejszego nazwanego kroku, dopóki publiczny klucz stanu lub klasa `RepeatCondition` będzie pasować.
`run_workflow( string $workflow_name, WorkflowBuilder $sub_builder )`	Dodaj krok uruchamiający przepływ pracy.
`delay( int $seconds = 0, int $minutes = 0, int $hours = 0, int $days = 0 )`	Dodaj trwały krok opóźnienia. Sumowane są wszystkie czasy trwania.
`wait_for_signal( string $name, ?string $result_key = null, ?string $step_name = null, array $match_payload = [], ?string $correlation_key = null )`	Dodaj krok oczekiwania na sygnał. Opcjonalne `match_payload` i `correlation_key` pozwalają ignorować niepowiązane zdarzenia.
`wait_for_signals( array $signal_names, WaitMode $mode = WaitMode::All, ?string $result_key = null, ?string $name = null, array $match_payload = [], ?string $correlation_key = null )`	Czekaj na wszystkie lub dowolne z kilku sygnałów, z opcjonalnym dopasowaniem ładunku i korelacją.
`wait_for_approval( string $signal_name = 'approval', ?string $result_key = 'approval', ?string $name = null, array $match_payload = [], ?string $correlation_key = null )`	Czekaj na sygnał zatwierdzenia przez człowieka.
`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 )`	Czekaj na zatwierdzenie lub odrzucenie i zapisz ustrukturyzowane dane decyzji.
`wait_for_input( string $signal_name = 'input', ?string $result_key = 'input', ?string $name = null, array $match_payload = [], ?string $correlation_key = null )`	Czekaj na sygnał wprowadzony przez człowieka.
`wait_for_workflow( int\|string $workflow, ?string $result_key = null, ?string $name = null )`	Czekaj na zakończenie jednego przepływu pracy o podanym ID lub ID zapisanego w stanie publicznym.
`wait_for_workflows( array\|string $workflows, WaitMode $mode = WaitMode::All, ?string $result_key = null, ?string $name = null, ?int $quorum = null )`	Czekaj na zakończenie wszystkich, dowolnych lub wymaganej liczby spośród kilku przepływów pracy.
`wait_for_workflow_group( string $group_key, WaitMode $mode = WaitMode::All, ?int $quorum = null, ?string $result_key = null, ?string $name = null )`	Czekaj na wcześniej uruchomioną grupę przepływów pracy, używając jej wewnętrznego klucza grupy.
`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 )`	Zleć jeden przepływ pracy najwyższego poziomu dla każdego elementu środowiska uruchomieniowego, opcjonalnie oznaczając uruchomiony zestaw trwałym kluczem grupy.
`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 dla zastosowań z agentami: `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 dla zastosowań z agentami: `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 dla zastosowań z agentami: `wait_for_workflow_group()`.
`version( string $version )`	Oznacz definicję przepływu pracy wersją na poziomie aplikacji, aby ułatwić jej przeglądanie i eksportowanie.
`idempotency_key( string $key )`	Zapewnij trwałość zlecenia przepływu pracy dzięki kluczowi podanemu przez wywołującego. Ponowne zlecenie z tym samym kluczem zwraca pierwotne ID przepływu pracy.
`max_transitions( int $max )`	Zakończ przepływ pracy niepowodzeniem, jeśli ukończono więcej niż `$max` przejść między krokami.
`max_for_each_items( int $max )`	Zakończ przepływ pracy niepowodzeniem, jeśli krok `for_each()` spróbuje utworzyć więcej niż `$max` elementów.
`max_state_bytes( int $bytes )`	Zakończ przepływ pracy niepowodzeniem, jeśli publiczny stan przekroczy `$bytes`.
`max_cost_units( int $max )`	Zakończ przepływ pracy niepowodzeniem, jeśli ukończone kroki zużyją więcej niż `$max` jednostek kosztu.
`max_started_workflows( int $max )`	Zakończ przepływ pracy niepowodzeniem, zanim zostanie zleconych więcej niż `$max` podrzędnych przepływów pracy.
`compensate_with( string $handler_class )`	Dołącz procedurę obsługi kompensacji do ostatnio dodanego kroku.
`compensate_on_failure()`	Uruchamiaj automatycznie kompensacje ukończonych kroków, gdy przepływ pracy zakończy się niepowodzeniem.
`on_queue( string $queue )`	Ustaw kolejkę dla wszystkich kroków.
`with_priority( Priority $priority )`	Ustaw priorytet dla wszystkich kroków.
`max_attempts( int $max )`	Ustaw maksymalną liczbę prób ponowienia dla każdego kroku.
`on_cancel( string $handler_class )`	Zarejestruj klasę procedury obsługi czyszczenia po anulowaniu. Zobacz sekcję Anulowanie.
`prune_state_after( int $steps = 2 )`	Włącz automatyczne przycinanie stanu. Usuwa wyniki kroków starsze niż określony próg. Zobacz sekcję Przycinanie stanu.
`must_complete_within( int $seconds = 0, int $minutes = 0, int $hours = 0, int $days = 0 )`	Ustaw termin wykonania przepływu pracy. Sumowane są wszystkie czasy trwania. Zobacz sekcję Terminy.
`on_deadline( string $handler_class )`	Zarejestruj procedurę obsługi wywoływaną po przekroczeniu terminu przepływu pracy. Zobacz sekcję Terminy.
`dispatch( array $initial_payload = [] )`	Zleć przepływ pracy. Zwraca jego ID.

`Queuety::workflow_status( $workflow_id )`

Pobierz status przepływu pracy. Zwraca obiekt WorkflowState lub 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 udostępnia również:

waiting_for: nazwy sygnałów lub identyfikatory przepływów pracy, które obecnie blokują wykonanie
current_step_name: skonfigurowana nazwa bieżącego kroku, jeśli została ustawiona
wait_mode: bieżący tryb oczekiwania (all, any lub quorum) podczas blokady
wait_details: dodatkowe dane do inspekcji, takie jak dopasowani i pozostali blokujący
definition_version: wartość ustawiona przez version(), jeśli została ustawiona
definition_hash: deterministyczny skrót definicji przepływu pracy
idempotency_key: trwały klucz dyspozycji, jeśli został ustawiony
budget: skonfigurowane limity oraz bieżące użycie, jeśli zabezpieczenia są włączone
artifact_count: liczba zapisanych artefaktów dla przepływu pracy
artifact_keys: klucze zapisanych artefaktów ułatwiające szybką inspekcję

`Queuety::retry_workflow( $workflow_id )`

Ponów nieudany przepływ pracy od kroku, w którym wystąpił błąd.

PHP

Queuety::retry_workflow( $workflow_id );

Zgłasza RuntimeException, jeśli po awarii przepływ pracy został już skompensowany.

`Queuety::pause_workflow( $workflow_id )`

Wstrzymaj działający przepływ pracy. Praca już zakolejkowana dla bieżącego kroku może się jeszcze zakończyć, ale kolejne kroki nie będą dodawane do kolejki, dopóki nie wznowisz przepływu pracy.

PHP

Queuety::pause_workflow( $workflow_id );

`Queuety::resume_workflow( $workflow_id )`

Wznów wstrzymany przepływ pracy.

PHP

Queuety::resume_workflow( $workflow_id );

`Queuety::cancel_workflow( $workflow_id )`

Anuluj przepływ pracy i uruchom wszystkie zarejestrowane procedury czyszczenia. Ustawia status na cancelled i przenosi wszystkie oczekujące oraz przetwarzane zadania przepływu pracy do archiwum.

PHP

Queuety::cancel_workflow( $workflow_id );

Zgłasza RuntimeException, jeśli przepływ pracy ma już status completed lub cancelled. Szczegóły znajdziesz w Anulowanie.

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

Cofnij przepływ pracy do poprzedniego kroku i ponownie uruchom go od tego miejsca. Stan zostanie przywrócony z dziennika zdarzeń state_after dla danego kroku, a następny krok zostanie dodany do kolejki.

PHP

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

Zgłasza RuntimeException, jeśli dla kroku docelowego nie istnieje zapisany stan. Wymaga Dziennika zdarzeń przepływu pracy. Szczegóły znajdziesz w Podróży w czasie.

`Queuety::fork_workflow( $workflow_id )`

Rozgałęź działający przepływ pracy, tworząc niezależną kopię w jego bieżącym stanie. Zwraca nowy identyfikator przepływu pracy.

PHP

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

Rozgałęziony przepływ pracy otrzymuje własny identyfikator i działa niezależnie od oryginału. Szczegóły znajdziesz w Rozgałęzianiu.

`Queuety::export_workflow( $workflow_id )`

Wyeksportuj pełną historię wykonania przepływu pracy jako tablicę możliwą do serializacji w JSON. Zawiera wiersz przepływu pracy, wszystkie zadania, zdarzenia przepływu pracy i wpisy dziennika.

PHP

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

Informacje o strukturze JSON i zastosowaniach znajdziesz w Eksportowaniu i odtwarzaniu.

`Queuety::replay_workflow( $data )`

Odtwórz wyeksportowany przepływ pracy w bieżącym środowisku. Tworzy nowy przepływ pracy na podstawie danych eksportu i zwraca jego nowy identyfikator.

PHP

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

Szczegóły znajdziesz w Eksportowaniu i odtwarzaniu.

`Queuety::replace_wp_cron()`

Zastąp wbudowany system WordPress cron harmonogramem narzędzia Queuety. Przechwytuje wp_schedule_event(), wp_unschedule_event() i wp_get_scheduled_event() oraz kieruje zdarzenia cykliczne do procesu roboczego Queuety.

PHP

Queuety::replace_wp_cron();

Działa tylko wtedy, gdy załadowano WordPress. Szczegóły znajdziesz w WP-Cron zastępowaniu.

`Queuety::restore_wp_cron()`

Przywróć domyślne działanie WordPress cron, usuwając zaczepy mostu Queuety cron.

PHP

Queuety::restore_wp_cron();

Informacje o ograniczeniach znajdziesz w Zastępowaniu WP-Cron.

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

Wyślij sygnał do przepływu pracy. Jeśli przepływ pracy oczekuje obecnie na ten sygnał, zostanie natychmiast wznowiony. W przeciwnym razie sygnał zostanie zapisany i obsłużony, gdy przepływ pracy dotrze do odpowiedniego wait_for_signal kroku.

PHP

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

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

Szczegóły znajdziesz w Sygnałach przepływu pracy.

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

Semantyczna nakładka na signal() dla przepływów zatwierdzania.

PHP

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

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

Semantyczna nakładka na signal() dla przepływów odrzucania.

PHP

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

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

Semantyczna nakładka na signal() dla ustrukturyzowanych danych wprowadzanych przez człowieka.

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

Zapisz jeden trwały artefakt dla przepływu pracy.

PHP

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

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

Zapisz jeden artefakt dla aktualnie wykonywanego kroku przepływu pracy.

PHP

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

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

Pobierz jeden zapisany artefakt dla przepływu pracy.

PHP

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

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

Wyświetl listę artefaktów przepływu pracy.

PHP

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

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

Usuń jeden artefakt przepływu pracy.

PHP

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

Maszyny stanów

`Queuety::machine( $name )`

Rozpocznij tworzenie maszyny stanów. Zwraca 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 ] );

Metody `StateMachineBuilder`

Metoda	Opis
`initial( string $state_name )`	Jawnie ustaw stan początkowy.
`state( string $state_name, ?StateMachineStatus $terminal_status = null )`	Rozpocznij konfigurację jednego stanu.
`action( string\|array $action_class )`	Zarejestruj zakolejkowaną akcję wejścia dla bieżącego stanu. Przekaż nazwę klasy lub ustrukturyzowaną definicję z `class`, `payload` i `config`; `payload` trafia do `StateAction::handle()` jako `$payload`.
`on( string $event, string $target, string\|array\|null $guard_class = null, ?string $name = null )`	Zarejestruj jedno przejście zdarzenia z bieżącego stanu. Warunek akceptuje tę samą ustrukturyzowaną postać co `action()`.
`on_queue( string $queue )`	Ustaw kolejkę używaną dla zakolejkowanych akcji stanów.
`with_priority( Priority $priority )`	Ustaw priorytet zakolejkowanych akcji stanów.
`max_attempts( int $max_attempts )`	Ustaw limit ponowień zakolejkowanych akcji stanów.
`version( string $version )`	Oznacz definicję maszyny wersją na poziomie aplikacji.
`idempotency_key( string $key )`	Zapewnij trwałość dyspozycji na podstawie klucza podanego przez wywołującego. Ponowne wywołanie z tym samym kluczem zwraca pierwotny identyfikator maszyny.
`dispatch( array $initial_state = [] )`	Uruchom dyspozycję maszyny i zwróć jej identyfikator.

Klasy akcji wejścia w stan mogą także udostępniać opcjonalną metodę config() z tymi samymi kluczami zasad dotyczących zasobów, które obsługują kroki przepływu pracy: concurrency_group, concurrency_limit i cost_units.

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

Uruchom serializowany pakiet definicji maszyny stanów. W większości kodu aplikacji lepiej użyć machine()->dispatch(). Ta metoda niższego poziomu pozwala adapterom uruchamiać maszynę bez bezpośredniej zależności od konstruktora.

PHP

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

`Queuety::machine_status( $machine_id )`

Pobierz bieżący status maszyny. Zwraca obiekt StateMachineState albo 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 udostępnia także:

definition_version
definition_hash
idempotency_key
error_message
current_action
terminal_status

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

Wyślij jedno zdarzenie zewnętrzne do oczekującej maszyny.

PHP

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

Dane zdarzenia są dostępne dla warunków przejść i po zaakceptowaniu zdarzenia są scalane z publicznym stanem maszyny.

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

Wyświetl zapisane wiersze maszyn do inspekcji.

PHP

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

Każdy wiersz zawiera identyfikator maszyny, nazwę, status, bieżący stan, wersję definicji, skrót definicji, klucz idempotencji i znaczniki czasu cyklu życia.

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

Pobierz trwałą oś czasu zdarzeń jednej maszyny.

PHP

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

Każdy wiersz zdarzenia zawiera: id, machine_id, job_id, state_name, handler, event, event_name, transition_name, queue, attempt, input, output, state_before, state_after, context, artifacts, chunks, error, duration_ms i created_at.

`Queuety::machine_trace( $machine_id )`

Pobierz znormalizowany pakiet śledzenia dla UI debuggera maszyny stanów.

PHP

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

Śledzenie zawiera machine, pogrupowane states, surowe events, skorelowaną akcję jobs, logs, artifacts i chunks.

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

Pobierz stan publiczny po określonym zdarzeniu śledzenia maszyny lub null, jeśli go nie znaleziono.

PHP

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

Model cyklu życia i wzorce użycia opisano w sekcji Maszyny stanów.

Szablony przepływów pracy

`Queuety::define_workflow( $name )`

Utwórz konstruktor przepływu pracy do użycia jako szablon.

PHP

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

`Queuety::register_workflow_template( $builder )`

Zarejestruj szablon przepływu pracy.

PHP

Queuety::register_workflow_template( $builder );

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

Uruchom zarejestrowany szablon przepływu pracy według nazwy. Zwraca identyfikator przepływu pracy.

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

Zgłasza RuntimeException, jeśli szablon nie jest zarejestrowany.

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

Uruchom przepływ pracy z pakietu definicji środowiska wykonawczego.

W większości kodu aplikacji lepiej użyć run_workflow() lub workflow()->dispatch(). Ta metoda niższego poziomu pozwala wielokrotnego użytku adapterom uruchamiać serializowaną definicję przepływu pracy bez zależności od wewnętrznych metod menedżera przepływów pracy.

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

Zarejestruj akcję WordPress, która uruchamia przepływ pracy.

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 może być jednym z następujących elementów:

nazwa zarejestrowanego szablonu przepływu pracy
wbudowany WorkflowBuilder

map, when i idempotency_key są wykonywane synchronicznie w ramach bieżącego żądania WordPress. Po uruchomieniu asynchronicznie wykonywane są tylko kroki przepływu pracy.

Pełny przewodnik znajdziesz w sekcji Wyzwalacze akcji.

`Queuety::workflow_templates()`

Pobierz rejestr szablonów przepływów pracy.

PHP

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

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

Wczytaj i zarejestruj wszystkie pliki przepływów pracy z katalogu. Każdy plik .php powinien zwracać WorkflowBuilder. Zwraca liczbę zarejestrowanych przepływów pracy.

PHP

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

`Queuety::load_workflow_file( $file_path )`

Wczytaj i zarejestruj pojedynczy plik przepływu pracy. Zwraca WorkflowTemplate lub null, jeśli plik nie zwraca konstruktora.

PHP

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

Harmonogramowanie

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

Utwórz harmonogram cykliczny. Zwraca konstruktor PendingSchedule.

PHP

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

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

Metody `PendingSchedule`

Metoda	Opis
`every( string $interval )`	Ustaw wyrażenie interwału (e.g. `1 hour`, `30 minutes`)
`cron( string $expression )`	Ustaw 5-polowe wyrażenie cron
`overlap( OverlapPolicy $policy )`	Ustaw zasady nakładania się (domyślnie: `Allow`). Zobacz sekcję Harmonogramowanie.
`on_queue( string $queue )`	Ustaw kolejkę docelową (domyślnie: `default`)
`id()`	Wymuś uruchomienie i zwróć identyfikator harmonogramu

`Queuety::scheduler()`

Pobierz wewnętrzną instancję Scheduler.

PHP

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

Rejestrowanie obsługi

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

Zarejestruj klasę obsługi pod określoną nazwą.

PHP

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

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

Automatycznie wykryj i zarejestruj klasy obsługi z katalogu. Zwraca liczbę zarejestrowanych klas obsługi.

PHP

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

`Queuety::registry()`

Pobierz rejestr klas obsługi.

PHP

$registry = Queuety::registry();

Obserwowalność

`Queuety::logger()`

Pobierz instancję Logger do odpytywania wpisów dziennika.

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

Pobierz instancję Metrics.

PHP

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

`Queuety::webhook_notifier()`

Pobierz instancję WebhookNotifier.

PHP

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

`Queuety::rate_limiter()`

Pobierz instancję RateLimiter.

PHP

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

Pamięć podręczna

`Queuety::cache()`

Pobierz instancję aktywnego zaplecza pamięci podręcznej.

PHP

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

`Queuety::set_cache( $cache )`

Zmień zaplecze pamięci podręcznej. Wywołaj przed Queuety::init(), aby użyć własnej implementacji.

PHP

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

`QUEUETY_CACHE_TTL`

Domyślny czas życia wpisów w pamięci podręcznej w sekundach (domyślnie: 5). Ustaw tę stałą w wp-config.php, aby zmienić domyślny czas życia używany przez wewnętrzne operacje pamięci podręcznej.

PHP

define( 'QUEUETY_CACHE_TTL', 10 );

Szczegóły dotyczące zapleczy i automatycznego wykrywania znajdziesz w sekcji Buforowanie.

Dziennik zdarzeń maszyny stanów

`Queuety::machine_events()`

Pobierz instancję StateMachineEventLog.

PHP

$log = Queuety::machine_events();

Dziennik zdarzeń przepływu pracy

`Queuety::workflow_events()`

Pobierz instancję WorkflowEventLog.

PHP

$log = Queuety::workflow_events();

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

Pobierz pełną oś czasu zdarzeń przepływu pracy. Zwraca tablicę wierszy zdarzeń uporządkowanych według identyfikatora.

PHP

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

Każdy wiersz zdarzenia zawiera: 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 (skopiowane z error.message), duration_ms i created_at.

`Queuety::workflow_trace( $workflow_id )`

Pobierz znormalizowany pakiet śledzenia dla UI debuggera przepływu pracy.

PHP

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

Śledzenie zawiera workflow, pogrupowane steps, surowe events, jobs, logs, artifacts, chunks, signals i wait_dependencies.

`Queuety::trace_input( $input )`

Ustaw dane wejściowe na poziomie aplikacji dla aktualnie wykonywanego kroku przepływu pracy.

`Queuety::trace_output( $output )`

Ustaw dane wyjściowe na poziomie aplikacji dla aktualnie wykonywanego kroku przepływu pracy.

`Queuety::trace_context( array $context )`

Scal kontekst na poziomie aplikacji z aktualnie wykonywanym krokiem przepływu pracy.

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

Pobierz stan publiczny po zakończeniu lub wznowieniu określonego kroku przepływu pracy albo null, jeśli go nie znaleziono.

PHP

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

Przykłady zastosowań i polecenia CLI opisano w sekcji Dziennik zdarzeń przepływu pracy.

Instancje wewnętrzne

Te metody zwracają instancje wewnętrzne do zaawansowanych zastosowań.

Metoda	Zwraca	Opis
`Queuety::queue()`	`Queue`	Operacje na kolejce (przejmowanie, zwalnianie, archiwizowanie)
`Queuety::worker()`	`Worker`	Sterowanie procesem roboczym
`Queuety::workflow_manager()`	`Workflow`	Orkiestracja przepływu pracy
`Queuety::machines()`	`StateMachine`	Zarządzanie cyklem życia maszyny stanów
`Queuety::connection()`	`Connection`	Połączenie z bazą danych
`Queuety::batch_manager()`	`BatchManager`	Zarządzanie cyklem życia partii (anulowanie, przycinanie)
`Queuety::chunk_store()`	`ChunkStore`	Utrwalanie fragmentów dla kroków strumieniowych
`Queuety::machine_events()`	`StateMachineEventLog`	Zdarzenia śledzenia maszyny stanów i znormalizowane pakiety śledzenia
`Queuety::workflow_events()`	`WorkflowEventLog`	Zdarzenia śledzenia przepływu pracy i znormalizowane pakiety śledzenia
`Queuety::resource_manager()`	`ResourceManager`	Profile dopuszczania procesów roboczych, ważone budżety i inspekcja pamięci systemowej
`Queuety::cache()`	`Cache`	Instancja zaplecza pamięci podręcznej

Testowanie

`Queuety::fake()`

Zastąp kolejkę atrapą działającą w pamięci na potrzeby testów. Zwraca instancję QueueFake, która rejestruje wysłane zadania bez dostępu do bazy danych. create_batch() rejestruje także metadane atrap partii; łańcuchy weryfikuje się za pomocą poszczególnych wysłanych zadań.

PHP

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

`Queuety::reset()`

Zresetuj cały stan wewnętrzny, w tym atrapę kolejki. Wywołaj w tearDown().

PHP

Queuety::reset();

Metody `QueueFake`

Metoda	Opis
`assert_pushed( string $class, ?Closure $callback )`	Sprawdź, czy wysłano klasę zadania. Opcjonalne wywołanie zwrotne filtruje według danych zadania.
`assert_pushed_times( string $class, int $count )`	Sprawdź, czy zadanie wysłano dokładnie N razy.
`assert_not_pushed( string $class )`	Sprawdź, czy zadania nie wysłano ani razu.
`assert_nothing_pushed()`	Sprawdź, czy nie wysłano żadnych zadań.
`assert_batched( ?Closure $callback )`	Sprawdź, czy wysłano partię. Opcjonalne wywołanie zwrotne filtruje według danych partii.
`pushed( string $class )`	Pobierz wszystkie zarejestrowane dyspozycje dla klasy. Zwraca tablicę.
`batches()`	Pobierz wszystkie zarejestrowane partie. Zwraca tablicę.
`reset()`	Wyczyść wszystkie zarejestrowane zadania i partie.

Pełne przykłady znajdziesz w sekcji Testowanie.

Interfejsy

`Handler`

Interfejs dla prostych klas obsługi zadań.

PHP

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

config() może zwracać: queue, max_attempts, backoff, rate_limit, concurrency_group, concurrency_limit i cost_units.

`Step`

Interfejs dla klas obsługi kroków przepływu pracy.

PHP

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

config() może zwracać: max_attempts, backoff, concurrency_group, concurrency_limit i cost_units.

`Contracts\ForEachHandler`

Interfejs dla klas obsługi dynamicznych gałęzi typu for-each.

PHP

namespace Queuety\Contracts;

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

config() może zwracać: max_attempts, backoff, rate_limit, concurrency_group, concurrency_limit i cost_units.

`Contracts\ForEachReducer`

Opcjonalny reduktor wyników for-each.

PHP

namespace Queuety\Contracts;

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

`Contracts\Compensation`

Interfejs dla klas obsługi kompensacji kroków.

PHP

namespace Queuety\Contracts;

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

`Contracts\RepeatCondition`

Interfejs predykatów pętli używanych przez repeat_until() i repeat_while(), gdy porównanie klucza stanu jest zbyt ograniczone.

PHP

namespace Queuety\Contracts;

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

`Contracts\StateAction`

Interfejs dla zakolejkowanych akcji wejścia w stan.

PHP

namespace Queuety\Contracts;

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

$payload przenosi ustrukturyzowane payload z definicji akcji, jeśli maszyna została tak zdefiniowana. Zwróć aktualizacje stanu publicznego. Uwzględnij _event i opcjonalnie _event_payload, jeśli akcja ma natychmiast wywołać przejście.

Akcje stanów mogą także definiować opcjonalną metodę config(). Queuety odczytuje z niej concurrency_group, concurrency_limit i cost_units, gdy akcja wejścia do stanu jest dodawana do kolejki.

`Contracts\StateGuard`

Interfejs dla warunków przejść maszyny stanów.

PHP

namespace Queuety\Contracts;

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

`Contracts\Job`

Interfejs dla samodzielnych klas zadań, które można wysyłać. Klasy zadań umieszczają dane w ładunku jako właściwości publiczne, a logikę wykonania w handle().

PHP

namespace Queuety\Contracts;

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

Klasy zadań mogą opcjonalnie definiować metodę middleware(), która zwraca tablicę instancji Contracts\Middleware.

`Contracts\Cache`

Interfejs dla zapleczy pamięci podręcznej. Wbudowane implementacje i własne zaplecza opisano w sekcji Buforowanie.

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

Metoda	Zwraca	Opis
`get( $key )`	`mixed`	Pobierz wartość. Zwraca `null`, jeśli jej nie znaleziono lub wygasła.
`set( $key, $value, $ttl )`	`void`	Zapisz wartość. TTL równy 0 oznacza brak wygaśnięcia.
`delete( $key )`	`void`	Usuń wartość.
`has( $key )`	`bool`	Sprawdź, czy klucz istnieje i nie wygasł.
`add( $key, $value, $ttl )`	`bool`	Atomowo ustaw wartość, jeśli nie istnieje. Zwraca `true`, jeśli wartość została ustawiona.
`flush()`	`void`	Usuń wszystkie elementy z pamięci podręcznej.

`WorkflowEventLog`

Zapisuje przejścia kroków przepływu pracy wraz z pełnym kontekstem śledzenia. Dostęp przez Queuety::workflow_events().

Metoda	Zwraca	Opis
`record_event( array $event )`	`void`	Zapisz surowe zdarzenie śledzenia
`record_step_started( ... )`	`void`	Zapisz zdarzenie step_started wraz z danymi wejściowymi i stanem przed wykonaniem
`record_step_completed( ... )`	`void`	Zapisz zdarzenie step_completed wraz z danymi wejściowymi, wyjściowymi, stanem przed i stanem po wykonaniu
`record_step_failed( ... )`	`void`	Zapisz zdarzenie step_failed wraz z kontekstem śledzenia i ustrukturyzowanymi szczegółami błędu
`get_timeline( int $workflow_id, ?int $limit = 100, int $offset = 0 )`	`array`	Pobierz zdarzenia śledzenia przepływu pracy w kolejności, z opcjonalną paginacją
`get_trace( int $workflow_id )`	`array`	Pobierz znormalizowany pakiet śledzenia
`get_state_at_step( int $workflow_id, int $step_index )`	`array\|null`	Pobierz stan publiczny po zakończeniu lub wznowieniu określonego kroku
`prune( int $older_than_days )`	`int`	Usuń zdarzenia starsze niż podana liczba dni. Zwraca liczbę usuniętych wierszy.

`StateMachineEventLog`

Zapisuje zdarzenia cyklu życia maszyny stanów, akcji, przejść i warunków, używając tych samych pól śledzenia co zdarzenia przepływu pracy.

Metoda	Zwraca	Opis
`record_event( array $event )`	`void`	Zapisuje surowe zdarzenie śledzenia maszyny
`get_timeline( int $machine_id, ?int $limit = 100, int $offset = 0 )`	`array`	Pobiera zdarzenia śledzenia maszyny w kolejności, z opcjonalnym stronicowaniem
`get_trace( int $machine_id )`	`array`	Pobiera znormalizowany pakiet śledzenia maszyny
`get_state_at_event( int $machine_id, int $event_id )`	`array\|null`	Pobiera stan publiczny po określonym zdarzeniu śledzenia maszyny
`prune( int $older_than_days )`	`int`	Usuwa zdarzenia starsze niż podana liczba dni. Zwraca liczbę usuniętych wierszy.

`Contracts\StreamingStep`

Interfejs dla obsługi kroków przepływu pracy, które generują strumień fragmentów. Każda zwrócona wartość jest natychmiast zapisywana w bazie danych. Zobacz Kroki strumieniowe.

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

Metoda	Opis
`stream( $state, $existing_chunks )`	Generator zwracający fragmenty ciągu znaków. Każde zwrócenie powoduje natychmiastowy zapis w bazie danych. Przy ponowieniu `$existing_chunks` zawiera wcześniej zapisane fragmenty.
`on_complete( $chunks, $state )`	Wywoływana po zakończeniu strumienia. Zwraca dane do połączenia ze stanem przepływu pracy.
`config()`	Opcjonalna konfiguracja. Obsługuje `max_attempts`, `backoff`, `concurrency_group`, `concurrency_limit` i `cost_units`.

`ChunkStore`

Zarządza trwałym zapisem fragmentów dla kroków strumieniowych. Dostęp przez Queuety::chunk_store().

Metoda	Zwraca	Opis
`get_chunks( int $job_id )`	`string[]`	Pobiera wszystkie fragmenty zadania, uporządkowane według indeksu fragmentu
`append_chunk( int $job_id, int $chunk_index, string $content, ?int $workflow_id, ?int $step_index )`	`void`	Dopisuje pojedynczy fragment
`clear_chunks( int $job_id )`	`void`	Usuwa wszystkie fragmenty zadania
`chunk_count( int $job_id )`	`int`	Zlicza fragmenty zadania
`get_accumulated( int $job_id )`	`string`	Łączy wszystkie fragmenty w jeden ciąg znaków

`Heartbeat`

Statyczna metoda pomocnicza do wysyłania sygnałów aktywności z poziomu obsługi kroków i kroków strumieniowych. Zobacz Sygnały aktywności.

Metoda	Opis
`Heartbeat::init( int $job_id, Connection $conn )`	Ustawia kontekst zadania. Wywoływana automatycznie przez proces roboczy.
`Heartbeat::beat( array $progress = [] )`	Wysyła sygnał aktywności. Aktualizuje `reserved_at` i opcjonalnie zapisuje dane o postępie w `heartbeat_data`.
`Heartbeat::clear()`	Czyści kontekst zadania. Wywoływana automatycznie przez proces roboczy.
`Heartbeat::current_job_id()`	Pobiera bieżący identyfikator zadania (do testów).

`Contracts\Middleware`

Interfejs oprogramowania pośredniczącego dla zadań. Oprogramowanie pośredniczące opakowuje wykonanie zadania w potok o strukturze cebuli.

PHP

namespace Queuety\Contracts;

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

Cechy

`Dispatchable`

Udostępnia statyczne metody dyspozytorskie w klasach zadań implementujących Contracts\Job. Argumenty konstruktora są przekazywane do utworzenia instancji zadania.

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

Metoda	Zwraca	Opis
`dispatch( ...$args )`	`PendingJob`	Tworzy instancję i wysyła ją do wykonania
`dispatch_if( bool $condition, ...$args )`	`PendingJob\|null`	Wysyła do wykonania tylko wtedy, gdy `$condition` ma wartość true
`dispatch_unless( bool $condition, ...$args )`	`PendingJob\|null`	Wysyła do wykonania tylko wtedy, gdy `$condition` ma wartość false
`with_chain( array $jobs )`	`ChainBuilder`	Rozpoczyna łańcuch od tej klasy zadania

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

Właściwości zadania

Klasy zadań mogą deklarować właściwości publiczne, które proces roboczy odczytuje za pomocą refleksji, aby nadpisać zachowanie dotyczące ponawiania prób i limitu czasu.

Właściwość	Typ	Opis
`$tries`	`int`	Maksymalna liczba prób przed umieszczeniem zadania w archiwum
`$timeout`	`int`	Maksymalny czas wykonania w sekundach (wymaga `pcntl`)
`$backoff`	`array`	Rosnące wartości opóźnienia w sekundach dla każdej kolejnej próby

PHP

class ImportProductsJob implements Job {
    use Dispatchable;

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

    // ...
}

Hak `failed()`

Gdy zadanie trafia do archiwum po wyczerpaniu wszystkich ponowień, proces roboczy wywołuje metodę failed() na instancji zadania, jeśli została zdefiniowana.

PHP

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

Wbudowane oprogramowanie pośredniczące

Wszystkie klasy oprogramowania pośredniczącego implementują Contracts\Middleware i znajdują się w przestrzeni nazw Queuety\Middleware.

Klasa	Opis
`RateLimited( int $max, int $window )`	Wymusza limit szybkości wynoszący `$max` wykonań na `$window` sekund
`Timeout( int $seconds )`	Kończy zadanie po `$seconds` sekundach (wymaga `pcntl`)
`UniqueJob( ?string $key )`	Zapobiega równoczesnemu wykonaniu za pomocą blokady bazy danych
`WithoutOverlapping( string $key, int $release_after )`	Zapobiega nakładającym się wykonaniom z automatycznym wygasaniem blokady
`ThrottlesExceptions( int $max_attempts, int $decay_minutes )`	Ogranicza liczbę wyjątków, aby zapobiec lawinie zadań, gdy usługa zewnętrzna nie działa

Szczegóły użycia i przykłady znajdziesz w sekcji Oprogramowanie pośredniczące.

Typy wyliczeniowe

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

Szczegółowe informacje o działaniu poszczególnych zasad znajdziesz w sekcji Planowanie.

Atrybuty

`#[QueuetyHandler]`

PHP 8 atrybut do automatycznej rejestracji. Zobacz PHP Atrybuty.

PHP

use Queuety\Attributes\QueuetyHandler;

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

Mostek CLI

`Queuety::cli_command()`

Zwraca opublikowaną nazwę głównego polecenia WP-CLI.

Domyślne polecenie główne to queuety. Zdefiniuj QUEUETY_CLI_COMMAND przed uruchomieniem Queuety, jeśli polecenie ma być opublikowane pod inną nazwą.

`Queuety::cli_command_map()`

Zwraca możliwy do serializacji katalog poleceń Queuety CLI dla środowisk testowych, które muszą poznać dostępne polecenia bez przeszukiwania klas WP-CLI.

Każdy wpis zawiera:

stabilny identyfikator operacji
względną ścieżkę CLI
klasę obsługi i metodę WP-CLI
możliwe cele wykonania
adapter normalizujący surowe argumenty CLI do planu wykonania

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

Przekształca przeanalizowane wywołanie CLI w plan wykonania, który powinno uruchomić środowisko testowe.

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

Plany wykonania uwzględniają używany transport:

polecenia są rozwiązywane do publicznych funkcji PHP ze znormalizowanymi argumentami
katalog z góry wymienia wszystkie możliwe cele wykonania, podczas gdy callable i transport wskazują gałąź wybraną dla przeanalizowanych danych wejściowych

To standardowe zastosowanie Queuety w środowiskach testowych agentów: stabilny identyfikator operacji, publiczny cel API oraz normalizacja przez adapter w kwestiach dotyczących wyłącznie CLI, takich jak domyślne kolejki, dekodowanie danych JSON, wybór gałęzi puli procesów roboczych i obsługa plików eksportu przepływu pracy.

PHP API

Dostęp do samodzielnego rdzenia

Queuety::init( Connection $conn )

Queuety::ensure_schema()

Zadania

Queuety::dispatch( $handler, $payload )

Metody PendingJob

Queuety::dispatch_sync( $handler, $payload )

Queuety::batch( $jobs )

Queuety::create_batch( $jobs )

Metody BatchBuilder

Queuety::find_batch( $id )

Sterowanie procesami roboczymi

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

Obiekt wartości Batch

Queuety::chain( $jobs )

Metody ChainBuilder

Queuety::retry( $job_id )

Queuety::retry_buried()

Queuety::purge( $older_than_days )

Kolejki

Queuety::stats( $queue )

Queuety::buried( $queue )

Queuety::pause( $queue )

Queuety::resume( $queue )

Queuety::is_paused( $queue )

Przepływy pracy

Queuety::workflow( $name )

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