PHP API

Als Markdown anzeigen Auf GitHub bearbeiten Dokumentation zu 2cc7b2b

Auf alles in Queuety kannst du über die statische Klasse Queuety\Queuety zugreifen. Sie ist der zentrale Einstiegspunkt, um Jobs auszuführen, Workflows zu erstellen, Zustandsautomaten zu steuern, Warteschlangen zu verwalten und auf interne Subsysteme zuzugreifen.

PHP

use Queuety\Queuety;

Eigenständiger Zugriff auf den Kern

Die DB-gestützte Laufzeit von Queuety lässt sich außerhalb von WordPress initialisieren.

Dieser eigenständige Pfad ist für den Orchestrierungskern gedacht:

Jobs und Worker (Worker-Pools benötigen außerdem die Konstanten DB_HOST, DB_NAME, DB_USER und DB_PASSWORD, siehe unten)
Workflows und Signale
Zustandsautomaten und Ereignisse
Artefakte, Logs, Metriken und Zeitpläne

Er ist kein Ersatz für den WordPress-Adapter. Queuety::on_action(), replace_wp_cron(), das Plugin-Bootstrap und der echte Registrierungspfad für wp queuety ... erfordern weiterhin WordPress.

`Queuety::init( Connection $conn )`

Initialisiere Queuety mit einer direkten DB-gestützten Verbindung.

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

Das optionale Argument table_prefix ändert den Queuety-Teil der Laufzeittabellennamen pro Verbindung:

PHP

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

Dadurch werden Tabellen wie queuety_jobs zu wp_themequeue_jobs aufgelöst.

`Queuety::ensure_schema()`

Stelle sicher, dass die Laufzeittabellen Queuety für die aktive Verbindung vorhanden sind.

PHP

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

Das vollständige Modell und die WordPress-Abgrenzung findest du unter Eigenständige Nutzung.

Jobs

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

Stelle einen Job in die Warteschlange. Akzeptiert entweder einen Handler-Namen als Zeichenfolge oder eine Instanz von Contracts\Job. Gibt einen PendingJob-Builder zurück, um Optionen anzuhängen.

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

Wird eine Instanz von Contracts\Job übergeben, extrahiert der Serializer den FQCN als Handler-Namen und öffentliche Eigenschaften als Nutzdaten. Der Parameter $payload wird ignoriert, wenn $handler eine Instanz von Job ist.

`PendingJob`-Methoden

Methode	Beschreibung
`on_queue( string $queue )`	Zielwarteschlange festlegen
`with_priority( Priority $priority )`	Prioritätsstufe festlegen
`delay( int $seconds )`	Verzögerung, bevor der Job verfügbar wird
`max_attempts( int $max )`	Maximale Anzahl der Wiederholungsversuche
`rate_limit( int $max, int $window )`	Ratenbegrenzung: maximale Ausführungen pro Zeitfenster in Sekunden
`unique()`	Doppelte Einreihungen verhindern
`after( int $job_id )`	Warten, bis zuerst ein anderer Job abgeschlossen ist
`concurrency_group( string $group, int $limit )`	Job einer gemeinsamen Worker-Zulassungsgruppe zuweisen
`cost_units( int $units )`	Der Einreihung Metadaten zu relativen Ausführungskosten hinzufügen
`id()`	Einreihung erzwingen und die Job-ID zurückgeben

concurrency_group() und cost_units() fließen in das umfassendere Modell für Ressourcenverwaltung ein. So kann dieselbe Einreihung an anzahlbasierten Parallelitätslimits, gewichteten Warteschlangen- oder Anbieterbudgets und Heuristiken zur Worker-Zulassung teilnehmen.

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

Führe einen Job synchron im aktuellen Prozess aus, ohne ihn in die Warteschlange aufzunehmen.

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

Die Methode handle() des Jobs wird direkt aufgerufen. Es wird kein Datenbankeintrag erstellt, kein Worker eingesetzt und keine Middleware angewendet.

`Queuety::batch( $jobs )`

Reihe mehrere Jobs mit einem einzelnen mehrzeiligen INSERT ein.

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

Erstelle ein BatchBuilder, um eine Gruppe von Jobs mit Callbacks und Fortschrittsverfolgung einzureihen. Ausführliche Informationen findest du unter Stapelverarbeitung.

PHP

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

`BatchBuilder`-Methoden

Methode	Rückgabe	Beschreibung
`name( string $name )`	`self`	Einen Namen für den Stapel festlegen
`on_queue( string $queue )`	`self`	Warteschlange für alle Jobs im Stapel festlegen
`then( string $handler_class )`	`self`	Handler, der bei erfolgreichem Abschluss des Stapels aufgerufen wird
`catch( string $handler_class )`	`self`	Handler, der aufgerufen wird, wenn der erste Job im Stapel fehlschlägt
`finally( string $handler_class )`	`self`	Handler, der beim Abschluss des Stapels aufgerufen wird (Erfolg oder Fehler)
`allow_failures()`	`self`	Callback `then` auch bei Fehlern ausführen lassen
`dispatch()`	`Batch`	Den Stapel-Datensatz und alle Jobs erstellen; gibt das Wertobjekt `Batch` zurück

`Queuety::find_batch( $id )`

Stapel anhand der ID suchen. Gibt ein Wertobjekt Batch oder null zurück.

PHP

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

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

Worker-Steuerung

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

Eine Worker-Schleife direkt ausführen.

PHP

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

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

Einen Worker-Pool mit fester Größe direkt ausführen. Erfordert die Erweiterungen pcntl und posix. Der Pool liest für seine Kindverbindungen die Konstanten WordPress DB_HOST, DB_NAME, DB_USER und DB_PASSWORD (nur die Tabellenpräfixe stammen von der initialisierten Verbindung). Definiere sie daher, wenn du einen Pool außerhalb von WordPress ausführst (#5).

PHP

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

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

Einen adaptiven Worker-Pool direkt ausführen. Es gelten dieselben Voraussetzungen wie für run_worker_pool().

PHP

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

Wertobjekt `Batch`

Methode/Eigenschaft	Typ	Beschreibung
`$id`	`int`	Stapel-ID
`$name`	`string\|null`	Optionaler Stapelname
`$total_jobs`	`int`	Gesamtzahl der Jobs im Stapel
`$pending_jobs`	`int`	Noch ausstehende Jobs
`$failed_jobs`	`int`	Anzahl fehlgeschlagener Jobs
`$failed_job_ids`	`array`	IDs fehlgeschlagener Jobs
`$created_at`	`DateTimeImmutable`	Zeitpunkt der Erstellung des Stapels
`$finished_at`	`DateTimeImmutable\|null`	Zeitpunkt des Abschlusses des Stapels
`$cancelled_at`	`DateTimeImmutable\|null`	Zeitpunkt der Abbruch des Stapels
`progress()`	`int`	Abschluss in Prozent (0-100)
`finished()`	`bool`	Gibt an, ob der Stapel abgeschlossen ist
`has_failures()`	`bool`	Gibt an, ob ein Job fehlgeschlagen ist
`cancelled()`	`bool`	Gibt an, ob der Stapel abgebrochen wurde

`Queuety::chain( $jobs )`

Erstelle ein ChainBuilder für die sequenzielle Job-Ausführung. Jeder Job hängt davon ab, dass der vorherige erfolgreich abgeschlossen wird.

PHP

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

`ChainBuilder`-Methoden

Methode	Rückgabe	Beschreibung
`on_queue( string $queue )`	`self`	Warteschlange für alle Jobs in der Kette festlegen
`catch( string $handler_class )`	`self`	Handler, der aufgerufen wird, wenn ein Job in der Kette endgültig archiviert wird
`dispatch()`	`int`	Alle Jobs mit sequenziellen Abhängigkeiten erstellen; gibt die ID des ersten Jobs zurück

`Queuety::retry( $job_id )`

Einen bestimmten Job erneut versuchen. Setzt ihn auf den Status pending zurück.

PHP

Queuety::retry( 42 );

`Queuety::retry_buried()`

Alle archivierten Jobs erneut versuchen. Gibt die Anzahl der erneut versuchten Jobs zurück.

PHP

$count = Queuety::retry_buried();

`Queuety::purge( $older_than_days )`

Abgeschlossene Jobs entfernen. Gibt die Anzahl der entfernten Jobs zurück.

PHP

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

Warteschlangen

`Queuety::stats( $queue )`

Job-Anzahlen nach Status gruppiert abrufen.

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

Alle archivierten Jobs abrufen.

PHP

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

`Queuety::pause( $queue )`

Eine Warteschlange pausieren, damit Worker sie überspringen.

PHP

Queuety::pause( 'emails' );

`Queuety::resume( $queue )`

Eine pausierte Warteschlange fortsetzen.

PHP

Queuety::resume( 'emails' );

`Queuety::is_paused( $queue )`

Prüfen, ob eine Warteschlange pausiert ist.

PHP

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

Workflows

`Queuety::workflow( $name )`

Workflow-Erstellung starten. Gibt ein WorkflowBuilder zurück.

PHP

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

`WorkflowBuilder`-Methoden

Methode	Beschreibung
`then( string $handler_class, ?string $name = null )`	Einen sequenziellen Schritt hinzufügen. Optionaler Name für bedingte Verzweigungen.
`parallel( array $handler_classes )`	Eine Gruppe paralleler Schritte hinzufügen.
`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 )`	Einen dynamischen For-each-Schritt hinzufügen, der für jedes Element im Workflow-Zustand einen Zweigjob erzeugt.
`repeat_while( string $target_step, ?string $state_key = null, mixed $expected = true, ?string $name = null, ?string $condition_class = null, ?int $max_iterations = null )`	Zu einem früheren benannten Schritt zurückspringen, solange ein öffentlicher Zustandsschlüssel oder eine Klasse `RepeatCondition` zutrifft.
`repeat_until( string $target_step, ?string $state_key = null, mixed $expected = true, ?string $name = null, ?string $condition_class = null, ?int $max_iterations = null )`	Zu einem früheren benannten Schritt zurückspringen, bis ein öffentlicher Zustandsschlüssel oder eine Klasse `RepeatCondition` zutrifft.
`run_workflow( string $workflow_name, WorkflowBuilder $sub_builder )`	Einen Workflow-Ausführungsschritt hinzufügen.
`delay( int $seconds = 0, int $minutes = 0, int $hours = 0, int $days = 0 )`	Einen dauerhaften Verzögerungsschritt hinzufügen. Alle Zeitspannen werden addiert.
`wait_for_signal( string $name, ?string $result_key = null, ?string $step_name = null, array $match_payload = [], ?string $correlation_key = null )`	Einen Signal-Warteschritt hinzufügen. Mit `match_payload` und `correlation_key` kannst du optional nicht zugehörige Ereignisse ignorieren.
`wait_for_signals( array $signal_names, WaitMode $mode = WaitMode::All, ?string $result_key = null, ?string $name = null, array $match_payload = [], ?string $correlation_key = null )`	Auf alle oder eines von mehreren Signalen warten, optional mit Nutzdatenabgleich und Korrelation.
`wait_for_approval( string $signal_name = 'approval', ?string $result_key = 'approval', ?string $name = null, array $match_payload = [], ?string $correlation_key = null )`	Auf ein menschliches Freigabesignal warten.
`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 )`	Auf Freigabe oder Ablehnung warten und eine strukturierte Entscheidungsnutzlast speichern.
`wait_for_input( string $signal_name = 'input', ?string $result_key = 'input', ?string $name = null, array $match_payload = [], ?string $correlation_key = null )`	Auf ein menschliches Eingabesignal warten.
`wait_for_workflow( int\|string $workflow, ?string $result_key = null, ?string $name = null )`	Auf den Abschluss einer Workflow-ID oder einer im öffentlichen Zustand gespeicherten Workflow-ID warten.
`wait_for_workflows( array\|string $workflows, WaitMode $mode = WaitMode::All, ?string $result_key = null, ?string $name = null, ?int $quorum = null )`	Auf den Abschluss aller, beliebiger oder einer ausreichenden Anzahl mehrerer Workflows warten.
`wait_for_workflow_group( string $group_key, WaitMode $mode = WaitMode::All, ?int $quorum = null, ?string $result_key = null, ?string $name = null )`	Über einen internen Gruppenschlüssel auf eine zuvor gestartete Workflow-Gruppe warten.
`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 )`	Einen Workflow der obersten Ebene pro Laufzeitelement einreihen und die gestartete Gruppe optional mit einem dauerhaften Gruppenschlüssel kennzeichnen.
`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 )`	Agentenorientierter Alias für `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 )`	Agentenorientierter Alias für `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 )`	Agentenorientierter Alias für `wait_for_workflow_group()`.
`version( string $version )`	Die Workflow-Definition für Prüfung und Exporte mit einer Anwendungsversion versehen.
`idempotency_key( string $key )`	Die Workflow-Einreihung für einen vom Aufrufer angegebenen Schlüssel dauerhaft machen. Eine erneute Einreihung mit demselben Schlüssel gibt die ursprüngliche Workflow-ID zurück.
`max_transitions( int $max )`	Workflow fehlschlagen lassen, wenn mehr als `$max` Schrittübergänge abgeschlossen werden.
`max_for_each_items( int $max )`	Workflow fehlschlagen lassen, wenn ein Schritt `for_each()` versucht, mehr als `$max` Elemente zu erzeugen.
`max_state_bytes( int $bytes )`	Workflow fehlschlagen lassen, wenn der öffentliche Zustand über `$bytes` hinauswächst.
`max_cost_units( int $max )`	Workflow fehlschlagen lassen, wenn abgeschlossene Schritte mehr als `$max` Kosteneinheiten verbrauchen.
`max_started_workflows( int $max )`	Workflow fehlschlagen lassen, bevor mehr als `$max` untergeordnete Workflows eingereiht werden.
`compensate_with( string $handler_class )`	Dem zuletzt hinzugefügten Schritt einen Kompensations-Handler zuweisen.
`compensate_on_failure()`	Kompensationen abgeschlossener Schritte automatisch ausführen, wenn der Workflow fehlschlägt.
`on_queue( string $queue )`	Warteschlange für alle Schritte festlegen.
`with_priority( Priority $priority )`	Priorität für alle Schritte festlegen.
`max_attempts( int $max )`	Maximale Anzahl der Wiederholungsversuche pro Schritt festlegen.
`on_cancel( string $handler_class )`	Eine Handler-Klasse für die Bereinigung bei Abbruch registrieren. Siehe Abbruch.
`prune_state_after( int $steps = 2 )`	Automatische Zustandsbereinigung aktivieren. Entfernt Schrittausgaben, die älter als der Schwellenwert sind. Siehe Zustandsbereinigung.
`must_complete_within( int $seconds = 0, int $minutes = 0, int $hours = 0, int $days = 0 )`	Eine Frist für den Workflow festlegen. Alle Zeitspannen werden addiert. Siehe Fristen.
`on_deadline( string $handler_class )`	Einen Handler registrieren, der aufgerufen wird, wenn die Workflow-Frist überschritten ist. Siehe Fristen.
`dispatch( array $initial_payload = [] )`	Workflow einreihen. Gibt die Workflow-ID zurück.

`Queuety::workflow_status( $workflow_id )`

Workflow-Status abrufen. Gibt ein Objekt WorkflowState oder null zurück.

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 stellt außerdem Folgendes bereit:

waiting_for: die Signalnamen oder Workflow-IDs, die den Lauf aktuell blockieren
current_step_name: der konfigurierte Name des aktuellen Schritts, falls vorhanden
wait_mode: der aktuelle Wartemodus (all, any oder quorum) während einer Blockierung
wait_details: zusätzliche Prüfdaten, etwa bereits erfüllte und verbleibende Blockierungen
definition_version: der von version() festgelegte Wert, falls vorhanden
definition_hash: ein deterministischer Hash der Workflow-Definition
idempotency_key: der dauerhafte Einreihungsschlüssel, falls vorhanden
budget: konfigurierte Limits sowie die aktuelle Nutzung, wenn Schutzvorkehrungen aktiviert sind
artifact_count: Anzahl der gespeicherten Artefakte für den Workflow
artifact_keys: gespeicherte Artefaktschlüssel zur schnellen Prüfung

`Queuety::retry_workflow( $workflow_id )`

Einen fehlgeschlagenen Workflow ab dem fehlgeschlagenen Schritt erneut versuchen.

PHP

Queuety::retry_workflow( $workflow_id );

Löst RuntimeException aus, wenn der Workflow nach einem Fehler bereits kompensiert wurde.

`Queuety::pause_workflow( $workflow_id )`

Einen laufenden Workflow pausieren. Bereits eingereihte Arbeit für den aktuellen Schritt kann noch abgeschlossen werden, spätere Schritte werden jedoch erst nach dem Fortsetzen des Workflows eingereiht.

PHP

Queuety::pause_workflow( $workflow_id );

`Queuety::resume_workflow( $workflow_id )`

Einen pausierten Workflow fortsetzen.

PHP

Queuety::resume_workflow( $workflow_id );

`Queuety::cancel_workflow( $workflow_id )`

Einen Workflow abbrechen und alle registrierten Bereinigungs-Handler ausführen. Setzt den Status auf cancelled und archiviert alle ausstehenden und laufenden Jobs des Workflows.

PHP

Queuety::cancel_workflow( $workflow_id );

Löst RuntimeException aus, wenn der Workflow bereits completed oder cancelled ist. Einzelheiten findest du unter Abbruch.

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

Einen Workflow auf einen vorherigen Schritt zurücksetzen und von dort erneut ausführen. Der Zustand wird anhand des Ereignisprotokolls state_after für den angegebenen Schritt wiederhergestellt, und der nächste Schritt wird eingereiht.

PHP

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

Löst RuntimeException aus, wenn für den Zielschritt kein gespeicherter Zustand vorhanden ist. Erfordert das Workflow-Ereignisprotokoll. Einzelheiten findest du unter Zeitreise.

`Queuety::fork_workflow( $workflow_id )`

Einen laufenden Workflow im aktuellen Zustand in eine unabhängige Kopie verzweigen. Gibt die neue Workflow-ID zurück.

PHP

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

Der verzweigte Workflow erhält eine eigene ID und läuft unabhängig vom Original. Einzelheiten findest du unter Verzweigen.

`Queuety::export_workflow( $workflow_id )`

Den vollständigen Ausführungsverlauf eines Workflows als JSON-serialisierbares Array exportieren. Enthält den Workflow-Datensatz, alle Jobs, Workflow-Ereignisse und Protokolleinträge.

PHP

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

Die Struktur JSON und Anwendungsfälle findest du unter Export und Wiedergabe.

`Queuety::replay_workflow( $data )`

Einen exportierten Workflow in der aktuellen Umgebung erneut ausführen. Erstellt anhand der Exportdaten einen neuen Workflow und gibt dessen neue ID zurück.

PHP

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

Einzelheiten findest du unter Export und Wiedergabe.

`Queuety::replace_wp_cron()`

Das integrierte WordPress-cron-System durch den Scheduler von Queuety ersetzen. Fängt wp_schedule_event(), wp_unschedule_event() und wp_get_scheduled_event() ab und leitet wiederkehrende Ereignisse über den Queuety-Worker weiter.

PHP

Queuety::replace_wp_cron();

Funktioniert nur, wenn WordPress geladen ist. Einzelheiten findest du unter WP-Cron-Ersatz.

`Queuety::restore_wp_cron()`

Das standardmäßige cron-Verhalten von WordPress wiederherstellen, indem die Bridge-Hooks Queuety cron entfernt werden.

PHP

Queuety::restore_wp_cron();

Einschränkungen findest du unter WP-Cron-Ersatz.

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

Ein Signal an einen Workflow senden. Wartet der Workflow gerade auf dieses Signal, wird er sofort fortgesetzt. Andernfalls wird das Signal gespeichert und abgerufen, sobald der Workflow den entsprechenden Schritt wait_for_signal erreicht.

PHP

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

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

Einzelheiten findest du unter Workflow-Signale.

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

Semantischer Wrapper um signal() für Freigabeabläufe.

PHP

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

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

Semantischer Wrapper um signal() für Ablehnungsabläufe.

PHP

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

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

Semantischer Wrapper um signal() für strukturierte menschliche Eingaben.

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

Ein dauerhaftes Artefakt für einen Workflow speichern.

PHP

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

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

Ein Artefakt für den aktuell ausgeführten Workflow-Schritt speichern.

PHP

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

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

Ein gespeichertes Artefakt für einen Workflow abrufen.

PHP

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

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

Artefakte für einen Workflow auflisten.

PHP

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

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

Ein Artefakt für einen Workflow löschen.

PHP

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

Zustandsautomaten

`Queuety::machine( $name )`

Erstellung eines Zustandsautomaten starten. Gibt ein StateMachineBuilder zurück.

PHP

use Queuety\Enums\StateMachineStatus;

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

`StateMachineBuilder`-Methoden

Methode	Beschreibung
`initial( string $state_name )`	Setzt den Anfangszustand ausdrücklich fest.
`state( string $state_name, ?StateMachineStatus $terminal_status = null )`	Beginnt mit der Konfiguration eines Zustands.
`action( string\|array $action_class )`	Registriert eine ausstehende Eintrittsaktion für den aktuellen Zustand. Übergib einen Klassennamen oder eine strukturierte Definition mit `class`, `payload` und `config`; die `payload` wird als `$payload` an `StateAction::handle()` übergeben.
`on( string $event, string $target, string\|array\|null $guard_class = null, ?string $name = null )`	Registriert einen Ereignisübergang vom aktuellen Zustand. Die Bedingung akzeptiert dieselbe strukturierte Form wie `action()`.
`on_queue( string $queue )`	Legt die Warteschlange für ausstehende Zustandsaktionen fest.
`with_priority( Priority $priority )`	Legt die Priorität für ausstehende Zustandsaktionen fest.
`max_attempts( int $max_attempts )`	Legt die Wiederholungsgrenze für ausstehende Zustandsaktionen fest.
`version( string $version )`	Kennzeichnet die Maschinendefinition mit einer Version auf Anwendungsebene.
`idempotency_key( string $key )`	Macht den Dispatch für einen vom Aufrufer angegebenen Schlüssel dauerhaft. Ein erneuter Dispatch mit demselben Schlüssel gibt dieselbe Maschinen-ID zurück.
`dispatch( array $initial_state = [] )`	Startet den Dispatch der Maschine und gibt die Maschinen-ID zurück.

Klassen für Zustands-Eintrittsaktionen können optional auch eine Methode config() mit denselben Ressourcenschlüsselwerten wie Workflow-Schritte bereitstellen: concurrency_group, concurrency_limit und cost_units.

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

Starte den Dispatch eines serialisierten Zustandsmaschinen-Definitionspakets. Die meisten Anwendungscodes sollten machine()->dispatch() bevorzugen. Diese Methode auf niedrigerer Ebene ermöglicht es Adaptern, eine Maschine zu starten, ohne direkt vom Builder abhängig zu sein.

PHP

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

`Queuety::machine_status( $machine_id )`

Ruft den aktuellen Maschinenstatus ab. Gibt ein StateMachineState-Objekt oder null zurück.

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 stellt außerdem Folgendes bereit:

definition_version
definition_hash
idempotency_key
error_message
current_action
terminal_status

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

Sendet ein externes Ereignis an eine wartende Maschine.

PHP

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

Die Nutzlast steht den Übergangsbedingungen zur Verfügung und wird bei Annahme des Ereignisses mit dem öffentlichen Maschinenzustand zusammengeführt.

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

Listet gespeicherte Maschinenzeilen zur Prüfung auf.

PHP

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

Jede Zeile enthält die Maschinen-ID, den Namen, den Status, den aktuellen Zustand, die Definitionsversion, den Definitions-Hash, den Idempotenzschlüssel und Zeitstempel des Lebenszyklus.

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

Ruft die dauerhafte Ereigniszeitleiste einer Maschine ab.

PHP

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

Jede Ereigniszeile enthält: 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 und created_at.

`Queuety::machine_trace( $machine_id )`

Ruft ein normalisiertes Trace-Paket für einen Zustandsmaschinen-Debugger UI ab.

PHP

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

Der Trace enthält machine, gruppierte states, rohe events, korrelierte Aktions-jobs, logs, artifacts und chunks.

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

Ruft den öffentlichen Zustand nach einem bestimmten Maschinen-Trace-Ereignis ab oder null, falls nichts gefunden wurde.

PHP

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

Weitere Informationen zum Lebenszyklusmodell und zu Verwendungsmustern findest du unter Zustandsmaschinen.

Workflow-Vorlagen

`Queuety::define_workflow( $name )`

Erstellt einen Workflow-Builder als Vorlage.

PHP

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

`Queuety::register_workflow_template( $builder )`

Registriert eine Workflow-Vorlage.

PHP

Queuety::register_workflow_template( $builder );

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

Startet den Dispatch einer registrierten Workflow-Vorlage anhand ihres Namens. Gibt die Workflow-ID zurück.

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

Wirft RuntimeException, wenn die Vorlage nicht registriert ist.

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

Startet den Dispatch eines Workflows aus einem Laufzeit-Definitionspaket.

Die meisten Anwendungscodes sollten run_workflow() oder workflow()->dispatch() bevorzugen. Diese Methode auf niedrigerer Ebene ermöglicht es wiederverwendbaren Adaptern, den Dispatch eines serialisierten Workflow-Definitionspakets zu starten, ohne von internen Workflow-Manager-Methoden abhängig zu sein.

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

Registriert eine WordPress-Aktion, die den Dispatch eines Workflows startet.

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 kann Folgendes sein:

der Name einer registrierten Workflow-Vorlage
ein Inline-WorkflowBuilder

map, when und idempotency_key werden synchron innerhalb der aktuellen WordPress-Anfrage ausgeführt. Nur die Workflow-Schritte selbst werden nach dem Dispatch asynchron ausgeführt.

Die vollständige Anleitung findest du unter Aktionsauslöser.

`Queuety::workflow_templates()`

Ruft die Workflow-Vorlagenregistrierung ab.

PHP

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

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

Lädt alle Workflow-Dateien aus einem Verzeichnis und registriert sie. Jede Datei vom Typ .php sollte ein WorkflowBuilder zurückgeben. Gibt die Anzahl der registrierten Workflows zurück.

PHP

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

`Queuety::load_workflow_file( $file_path )`

Lädt eine einzelne Workflow-Datei und registriert sie. Gibt WorkflowTemplate zurück oder null, wenn die Datei keinen Builder zurückgibt.

PHP

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

Zeitplanung

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

Erstellt einen wiederkehrenden Zeitplan. Gibt einen PendingSchedule-Builder zurück.

PHP

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

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

Methoden von `PendingSchedule`

Methode	Beschreibung
`every( string $interval )`	Legt einen Intervall-Ausdruck fest (e.g. `1 hour`, `30 minutes`)
`cron( string $expression )`	Legt einen 5-Feld-cron-Ausdruck fest
`overlap( OverlapPolicy $policy )`	Legt die Richtlinie für Überlappungen fest (Standard: `Allow`). Siehe Zeitplanung.
`on_queue( string $queue )`	Legt die Zielwarteschlange fest (Standard: `default`)
`id()`	Erzwingt den Dispatch und gibt die Zeitplan-ID zurück

`Queuety::scheduler()`

Ruft die interne Scheduler-Instanz ab.

PHP

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

Handler-Registrierung

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

Registriert eine Handler-Klasse unter einem Namen.

PHP

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

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

Erkennt Handler-Klassen in einem Verzeichnis automatisch und registriert sie. Gibt die Anzahl der registrierten Handler zurück.

PHP

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

`Queuety::registry()`

Ruft die Handler-Registrierung ab.

PHP

$registry = Queuety::registry();

Beobachtbarkeit

`Queuety::logger()`

Ruft die Logger-Instanz zum Abfragen von Protokolleinträgen ab.

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

Ruft die Metrics-Instanz ab.

PHP

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

`Queuety::webhook_notifier()`

Ruft die WebhookNotifier-Instanz ab.

PHP

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

`Queuety::rate_limiter()`

Ruft die RateLimiter-Instanz ab.

PHP

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

Cache

`Queuety::cache()`

Ruft die aktive Cache-Backend-Instanz ab.

PHP

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

`Queuety::set_cache( $cache )`

Setzt das Cache-Backend außer Kraft. Rufe diese Methode vor Queuety::init() auf, um eine eigene Implementierung zu verwenden.

PHP

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

`QUEUETY_CACHE_TTL`

Standard-TTL des Caches in Sekunden (Standard: 5). Setze diese Konstante in wp-config.php, um die Standard-TTL für interne Cache-Vorgänge zu ändern.

PHP

define( 'QUEUETY_CACHE_TTL', 10 );

Details zu Backends und zur automatischen Erkennung findest du unter Caching.

Ereignisprotokoll der Zustandsmaschine

`Queuety::machine_events()`

Ruft die StateMachineEventLog-Instanz ab.

PHP

$log = Queuety::machine_events();

Workflow-Ereignisprotokoll

`Queuety::workflow_events()`

Ruft die WorkflowEventLog-Instanz ab.

PHP

$log = Queuety::workflow_events();

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

Ruft die vollständige Ereigniszeitleiste eines Workflows ab. Gibt ein Array von Ereigniszeilen zurück, sortiert nach ID.

PHP

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

Jede Ereigniszeile enthält: 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 (kopiert aus error.message), duration_ms und created_at.

`Queuety::workflow_trace( $workflow_id )`

Ruft ein normalisiertes Trace-Paket für einen Workflow-Debugger UI ab.

PHP

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

Der Trace enthält workflow, gruppierte steps, rohe events, jobs, logs, artifacts, chunks, signals und wait_dependencies.

`Queuety::trace_input( $input )`

Legt anwendungsbezogene Eingaben für den gerade ausgeführten Workflow-Schritt fest.

`Queuety::trace_output( $output )`

Legt anwendungsbezogene Ausgaben für den gerade ausgeführten Workflow-Schritt fest.

`Queuety::trace_context( array $context )`

Führt anwendungsbezogenen Kontext mit dem gerade ausgeführten Workflow-Schritt zusammen.

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

Ruft den öffentlichen Zustand ab, nachdem ein bestimmter Workflow-Schritt abgeschlossen oder fortgesetzt wurde, oder null, falls nichts gefunden wurde.

PHP

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

Anwendungsfälle und CLI-Befehle findest du unter Workflow-Ereignisprotokoll.

Interne Instanzen

Diese Methoden geben interne Instanzen für fortgeschrittene Anwendungsfälle zurück.

Methode	Rückgabe	Beschreibung
`Queuety::queue()`	`Queue`	Warteschlangenoperationen (Beanspruchen, Freigeben, Verbergen)
`Queuety::worker()`	`Worker`	Steuerung von Worker-Prozessen
`Queuety::workflow_manager()`	`Workflow`	Workflow-Orchestrierung
`Queuety::machines()`	`StateMachine`	Verwaltung des Lebenszyklus von Zustandsmaschinen
`Queuety::connection()`	`Connection`	Datenbankverbindung
`Queuety::batch_manager()`	`BatchManager`	Verwaltung des Batch-Lebenszyklus (Abbrechen, Bereinigen)
`Queuety::chunk_store()`	`ChunkStore`	Speicherung von Chunks für Streaming-Schritte
`Queuety::machine_events()`	`StateMachineEventLog`	Trace-Ereignisse für Zustandsmaschinen und normalisierte Trace-Pakete
`Queuety::workflow_events()`	`WorkflowEventLog`	Workflow-Trace-Ereignisse und normalisierte Trace-Pakete
`Queuety::resource_manager()`	`ResourceManager`	Worker-Zulassungsprofile, gewichtete Budgets und Prüfung des Systemspeichers
`Queuety::cache()`	`Cache`	Cache-Backend-Instanz

Testen

`Queuety::fake()`

Ersetzt die Warteschlange zum Testen durch ein In-Memory-Fake. Gibt eine QueueFake-Instanz zurück, die gestartete Jobs protokolliert, ohne auf die Datenbank zuzugreifen. create_batch() protokolliert auch Fake-Batch-Metadaten; Chains werden anhand der einzelnen von ihnen gestarteten Jobs geprüft.

PHP

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

`Queuety::reset()`

Setzt den gesamten internen Zustand zurück, einschließlich des Warteschlangen-Fakes. Rufe die Methode in tearDown() auf.

PHP

Queuety::reset();

Methoden von `QueueFake`

Methode	Beschreibung
`assert_pushed( string $class, ?Closure $callback )`	Prüft, ob eine Job-Klasse in die Warteschlange gestellt wurde. Ein optionaler Callback filtert nach Jobdaten.
`assert_pushed_times( string $class, int $count )`	Prüft, ob ein Job genau N-mal in die Warteschlange gestellt wurde.
`assert_not_pushed( string $class )`	Prüft, ob ein Job nie in die Warteschlange gestellt wurde.
`assert_nothing_pushed()`	Prüft, ob überhaupt keine Jobs in die Warteschlange gestellt wurden.
`assert_batched( ?Closure $callback )`	Prüft, ob ein Batch gestartet wurde. Ein optionaler Callback filtert nach Batchdaten.
`pushed( string $class )`	Ruft alle aufgezeichneten Dispatches für eine Klasse ab. Gibt ein Array zurück.
`batches()`	Ruft alle aufgezeichneten Batches ab. Gibt ein Array zurück.
`reset()`	Löscht alle aufgezeichneten Jobs und Batches.

Vollständige Beispiele findest du unter Testen.

Schnittstellen

`Handler`

Schnittstelle für einfache Job-Handler.

PHP

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

config() kann queue, max_attempts, backoff, rate_limit, concurrency_group, concurrency_limit und cost_units zurückgeben.

`Step`

Schnittstelle für Workflow-Schritt-Handler.

PHP

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

config() kann max_attempts, backoff, concurrency_group, concurrency_limit und cost_units zurückgeben.

`Contracts\ForEachHandler`

Schnittstelle für dynamische Handler für For-Each-Verzweigungen.

PHP

namespace Queuety\Contracts;

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

config() kann max_attempts, backoff, rate_limit, concurrency_group, concurrency_limit und cost_units zurückgeben.

`Contracts\ForEachReducer`

Optionaler Reduzierer für For-Each-Ergebnisse.

PHP

namespace Queuety\Contracts;

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

`Contracts\Compensation`

Schnittstelle für Handler zur Kompensation von Schritten.

PHP

namespace Queuety\Contracts;

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

`Contracts\RepeatCondition`

Schnittstelle für Schleifenbedingungen, die von repeat_until() und repeat_while() verwendet werden, wenn ein Vergleich anhand eines Zustandsschlüssels zu eingeschränkt ist.

PHP

namespace Queuety\Contracts;

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

`Contracts\StateAction`

Schnittstelle für ausstehende Zustands-Eintrittsaktionen.

PHP

namespace Queuety\Contracts;

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

$payload erhält die strukturierte payload aus der Aktionsdefinition, sofern die Maschine damit definiert wurde. Gibt Aktualisierungen des öffentlichen Zustands zurück. Füge _event und optional _event_payload hinzu, wenn die Aktion sofort einen Übergang auslösen soll.

Zustandsaktionen können auch eine optionale Methode config() definieren. Queuety liest daraus concurrency_group, concurrency_limit und cost_units, wenn die Eintrittsaktion in die Warteschlange gestellt wird.

`Contracts\StateGuard`

Schnittstelle für Übergangsbedingungen von Zustandsmaschinen.

PHP

namespace Queuety\Contracts;

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

`Contracts\Job`

Schnittstelle für eigenständige Job-Klassen, die sich dispatchen lassen. Job-Klassen kapseln ihre Nutzlast in öffentlichen Eigenschaften und ihre Ausführungslogik in handle().

PHP

namespace Queuety\Contracts;

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

Job-Klassen können optional eine Methode middleware() definieren, die ein Array von Contracts\Middleware-Instanzen zurückgibt.

`Contracts\Cache`

Schnittstelle für Cache-Backends. Integrierte Implementierungen und eigene Backends findest du unter Caching.

PHP

namespace Queuety\Contracts;

interface Cache {
    public function get( string $key ): mixed;
    public function set( string $key, mixed $value, int $ttl = 0 ): void;
    public function delete( string $key ): void;
    public function has( string $key ): bool;
    public function add( string $key, mixed $value, int $ttl = 0 ): bool;
    public function flush(): void;
}

Methode	Rückgabe	Beschreibung
`get( $key )`	`mixed`	Ruft einen Wert ab. Gibt `null` zurück, wenn der Wert nicht gefunden wurde oder abgelaufen ist.
`set( $key, $value, $ttl )`	`void`	Speichert einen Wert. Eine TTL von 0 bedeutet, dass er nicht abläuft.
`delete( $key )`	`void`	Entfernt einen Wert.
`has( $key )`	`bool`	Prüft, ob ein Schlüssel existiert und nicht abgelaufen ist.
`add( $key, $value, $ttl )`	`bool`	Atomare Aktion zum Setzen, falls nicht vorhanden. Gibt `true` zurück, wenn der Wert gesetzt wurde.
`flush()`	`void`	Entfernt alle Elemente aus dem Cache.

`WorkflowEventLog`

Protokolliert Workflow-Schrittübergänge mit vollständigem Trace-Kontext. Zugriff über Queuety::workflow_events().

Methode	Rückgabe	Beschreibung
`record_event( array $event )`	`void`	Zeichnet ein rohes Trace-Ereignis auf
`record_step_started( ... )`	`void`	Zeichnet ein step_started-Ereignis mit Eingabe und Zustand vor der Ausführung auf
`record_step_completed( ... )`	`void`	Zeichnet ein step_completed-Ereignis mit Eingabe, Ausgabe, Zustand davor und Zustand danach auf
`record_step_failed( ... )`	`void`	Zeichnet ein step_failed-Ereignis mit Trace-Kontext und strukturierten Fehlerdetails auf
`get_timeline( int $workflow_id, ?int $limit = 100, int $offset = 0 )`	`array`	Ruft Workflow-Trace-Ereignisse der Reihe nach ab, optional mit Seitennummerierung
`get_trace( int $workflow_id )`	`array`	Ruft ein normalisiertes Trace-Paket ab
`get_state_at_step( int $workflow_id, int $step_index )`	`array\|null`	Ruft den öffentlichen Zustand nach Abschluss oder Fortsetzung eines bestimmten Schritts ab
`prune( int $older_than_days )`	`int`	Löscht Ereignisse, die älter als die angegebene Anzahl an Tagen sind. Gibt die Anzahl der gelöschten Zeilen zurück.

`StateMachineEventLog`

Protokolliert Lebenszyklus-, Aktions-, Übergangs- und Bedingungsereignisse von Zustandsmaschinen mit denselben Trace-Feldern wie Workflow-Ereignisse.

Methode	Rückgabe	Beschreibung
`record_event( array $event )`	`void`	Zeichnet ein rohes Maschinen-Trace-Ereignis auf
`get_timeline( int $machine_id, ?int $limit = 100, int $offset = 0 )`	`array`	Ruft Maschinen-Trace-Ereignisse der Reihe nach ab, optional mit Seitennummerierung
`get_trace( int $machine_id )`	`array`	Ruft ein normalisiertes Maschinen-Trace-Paket ab
`get_state_at_event( int $machine_id, int $event_id )`	`array\|null`	Ruft den öffentlichen Zustand nach einem bestimmten Maschinen-Trace-Ereignis ab
`prune( int $older_than_days )`	`int`	Löscht Ereignisse, die älter als die angegebene Anzahl an Tagen sind. Gibt die Anzahl der gelöschten Zeilen zurück.

`Contracts\StreamingStep`

Schnittstelle für Workflow-Schritt-Handler, die einen Stream von Chunks ausgeben. Jeder ausgegebene Wert wird sofort in der Datenbank gespeichert. Siehe Streaming-Schritte.

PHP

namespace Queuety\Contracts;

interface StreamingStep {
    public function stream( array $state, array $existing_chunks = [] ): \Generator;
    public function on_complete( array $chunks, array $state ): array;
    public function config(): array;
}

Methode	Beschreibung
`stream( $state, $existing_chunks )`	Generator, der String-Chunks ausgibt. Jedes Ausgeben löst sofort einen Schreibvorgang in der DB aus. Bei einem erneuten Versuch enthält `$existing_chunks` zuvor gespeicherte Chunks.
`on_complete( $chunks, $state )`	Wird nach Ende des Streams aufgerufen. Gibt Daten zurück, die mit dem Workflow-Zustand zusammengeführt werden.
`config()`	Optionale Konfiguration. Unterstützt `max_attempts`, `backoff`, `concurrency_group`, `concurrency_limit` und `cost_units`.

`ChunkStore`

Verwaltet die Speicherung von Chunks für Streaming-Schritte. Zugriff über Queuety::chunk_store().

Methode	Rückgabe	Beschreibung
`get_chunks( int $job_id )`	`string[]`	Ruft alle Chunks eines Jobs ab, sortiert nach Chunk-Index
`append_chunk( int $job_id, int $chunk_index, string $content, ?int $workflow_id, ?int $step_index )`	`void`	Hängt einen einzelnen Chunk an
`clear_chunks( int $job_id )`	`void`	Löscht alle Chunks eines Jobs
`chunk_count( int $job_id )`	`int`	Zählt die Chunks eines Jobs
`get_accumulated( int $job_id )`	`string`	Verbindet alle Chunks zu einem String

`Heartbeat`

Statische Hilfsfunktion zum Senden von Aktivitäts-Heartbeats innerhalb von Schritt-Handlern und Streaming-Schritten. Siehe Heartbeats.

Methode	Beschreibung
`Heartbeat::init( int $job_id, Connection $conn )`	Setzt den Job-Kontext. Wird automatisch vom Worker aufgerufen.
`Heartbeat::beat( array $progress = [] )`	Sendet einen Heartbeat. Aktualisiert `reserved_at` und speichert optional Fortschrittsdaten in `heartbeat_data`.
`Heartbeat::clear()`	Löscht den Job-Kontext. Wird automatisch vom Worker aufgerufen.
`Heartbeat::current_job_id()`	Ruft die aktuelle Job-ID ab (zum Testen).

`Contracts\Middleware`

Schnittstelle für Job-Middleware. Middleware umschließt die Job-Ausführung in einer Pipeline im Zwiebelstil.

PHP

namespace Queuety\Contracts;

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

Traits

`Dispatchable`

Stellt statische Dispatch-Methoden für Job-Klassen bereit, die Contracts\Job implementieren. Konstruktorargumente werden zum Erstellen der Job-Instanz übergeben.

PHP

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

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

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

Methode	Rückgabe	Beschreibung
`dispatch( ...$args )`	`PendingJob`	Erstellt eine Instanz und startet ihren Dispatch
`dispatch_if( bool $condition, ...$args )`	`PendingJob\|null`	Startet den Dispatch nur, wenn `$condition` wahr ist
`dispatch_unless( bool $condition, ...$args )`	`PendingJob\|null`	Startet den Dispatch nur, wenn `$condition` falsch ist
`with_chain( array $jobs )`	`ChainBuilder`	Startet eine Chain aus dieser Job-Klasse

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

Job-Eigenschaften

Job-Klassen können öffentliche Eigenschaften deklarieren, die der Worker per Reflection ausliest, um Wiederholungs- und Timeout-Verhalten anzupassen.

Eigenschaft	Typ	Beschreibung
`$tries`	`int`	Maximale Anzahl an Versuchen, bevor der Job verborgen wird
`$timeout`	`int`	Maximale Ausführungszeit in Sekunden (erfordert `pcntl`)
`$backoff`	`array`	Zunehmende Verzögerungswerte in Sekunden für jeden Wiederholungsversuch

PHP

class ImportProductsJob implements Job {
    use Dispatchable;

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

    // ...
}

Hook `failed()`

Wenn ein Job nach Ausschöpfen aller Wiederholungsversuche verborgen wird, ruft der Worker eine Methode failed() für die Job-Instanz auf, sofern sie definiert ist.

PHP

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

Integrierte Middleware

Alle Middleware-Klassen implementieren Contracts\Middleware und befinden sich im Namespace Queuety\Middleware.

Klasse	Beschreibung
`RateLimited( int $max, int $window )`	Begrenzt die Rate auf `$max` Ausführungen pro `$window` Sekunden
`Timeout( int $seconds )`	Beendet den Job nach `$seconds` Sekunden (erfordert `pcntl`)
`UniqueJob( ?string $key )`	Verhindert gleichzeitige Ausführung durch eine Datenbanksperre
`WithoutOverlapping( string $key, int $release_after )`	Verhindert überlappende Ausführung mit automatischem Ablauf der Sperre
`ThrottlesExceptions( int $max_attempts, int $decay_minutes )`	Drosselt Ausnahmen, um Job-Stürme bei Ausfall eines externen Dienstes zu verhindern

Weitere Informationen zur Verwendung und Beispiele findest du unter Middleware.

Aufzählungstypen

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

Details zur Funktionsweise der einzelnen Richtlinien findest du unter Scheduling.

Attribute

`#[QueuetyHandler]`

PHP-Attribut 8 für die automatische Registrierung. Siehe PHP Attributes.

PHP

use Queuety\Attributes\QueuetyHandler;

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

CLI-Bridge

`Queuety::cli_command()`

Gibt den veröffentlichten WP-CLI-Root-Befehlsnamen zurück.

Der Standard-Root-Befehl lautet queuety. Definiere QUEUETY_CLI_COMMAND, bevor Queuety gestartet wird, wenn der Befehl unter einem anderen Namen veröffentlicht werden soll.

`Queuety::cli_command_map()`

Gibt den serialisierbaren Queuety-CLI-Katalog für Harnesses zurück, die die bereitgestellte Befehlsoberfläche verstehen müssen, ohne WP-CLI-Klassen auszulesen.

Jeder Eintrag enthält:

eine stabile Operations-ID
den relativen CLI-Pfad
die WP-CLI-Handlerklasse und -Methode
die möglichen Ausführungsziele
den Adapter, der rohe CLI-Argumente in einen Ausführungsplan normalisiert

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

Wandelt einen geparsten CLI-Aufruf in den Ausführungsplan um, den das Harness ausführen soll.

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

Ausführungspläne berücksichtigen bewusst den Transport:

Befehle werden zu öffentlichen PHP-Aufrufobjekten mit normalisierten Argumenten aufgelöst
Der Katalog führt alle möglichen Ausführungsziele vorab auf, während callable und transport den für die geparste Eingabe gewählten Zweig kennzeichnen

Dies ist die Standardverwendung von Queuety für Agenten-Harnesses: stabile Operations-ID, öffentliches API-Ziel und adapterseitige Normalisierung für nur CLI betreffende Aspekte wie Queue-Standardwerte, JSON-Payload-Dekodierung, Worker-Pool-Verzweigungen und den Umgang mit Workflow-Exportdateien.

PHP API

Eigenständiger Zugriff auf den Kern

Queuety::init( Connection $conn )

Queuety::ensure_schema()

Jobs

Queuety::dispatch( $handler, $payload )

PendingJob-Methoden

Queuety::dispatch_sync( $handler, $payload )

Queuety::batch( $jobs )

Queuety::create_batch( $jobs )

BatchBuilder-Methoden

Queuety::find_batch( $id )

Worker-Steuerung

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

Wertobjekt Batch

Queuety::chain( $jobs )

ChainBuilder-Methoden

Queuety::retry( $job_id )

Queuety::retry_buried()

Queuety::purge( $older_than_days )

Warteschlangen

Queuety::stats( $queue )

Queuety::buried( $queue )

Queuety::pause( $queue )

Queuety::resume( $queue )

Queuety::is_paused( $queue )

Workflows

Queuety::workflow( $name )

WorkflowBuilder-Methoden

Queuety::workflow_status( $workflow_id )

Queuety::retry_workflow( $workflow_id )

Queuety::pause_workflow( $workflow_id )

Queuety::resume_workflow( $workflow_id )

Queuety::cancel_workflow( $workflow_id )

Queuety::rewind_workflow( $workflow_id, $to_step )

Queuety::fork_workflow( $workflow_id )

Queuety::export_workflow( $workflow_id )

Queuety::replay_workflow( $data )

Queuety::replace_wp_cron()

Queuety::restore_wp_cron()

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

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

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

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

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

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

Queuety::workflow_artifact( $workflow_id, $artifact_key )

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

Queuety::delete_workflow_artifact( $workflow_id, $artifact_key )