Przejdź do treści

Sygnały workflow

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

Workflow może wstrzymać działanie i oczekiwać na dane zewnętrzne za pomocą sygnałów. Metody konstruktora wait_for_signal() i wait_for_signals() dodają kroki blokujące workflow do chwili, gdy aplikacja, operator lub inny system wyśle oczekiwany sygnał za pośrednictwem Queuety::signal().

W przypadku procesów z udziałem człowieka Queuety obejmuje również:

  • wait_for_approval() dla zanonimizowanego ładunku zatwierdzenia
  • wait_for_input() dla uporządkowanych danych od człowieka
  • wait_for_decision() dla jawnych wyników zatwierdzenia lub odrzucenia

Większy przykład kompleksowego procesu łączącego uruchomienia agentów z etapami zatwierdzania przez człowieka znajdziesz w sekcji Orkiestracja agentów.

Metoda ->wait_for_signal()

Dodaj krok oczekiwania na sygnał do workflow:

PHP
use Queuety\Queuety;

$workflow_id = Queuety::workflow( 'content_approval' )
    ->then( DraftContentHandler::class )
    ->wait_for_signal( 'approved' )
    ->then( PublishContentHandler::class )
    ->dispatch( [ 'post_id' => 99 ] );

Gdy workflow dojdzie do kroku wait_for_signal, jego status zmieni się na waiting_for_signal i workflow zostanie wstrzymany. Podczas oczekiwania nie są uruchomione żadne zadania ani wykorzystywane żadne zasoby.

Domyślnie pojedynczy ładunek sygnału jest łączony z głównym stanem workflow. Jeśli zamiast tego chcesz umieścić ładunek w osobnej przestrzeni nazw, przekaż result_key:

PHP
$workflow_id = Queuety::workflow( 'content_approval' )
    ->then( DraftContentHandler::class )
    ->wait_for_signal( 'approved', 'approval' )
    ->then( PublishContentHandler::class )
    ->dispatch( [ 'post_id' => 99 ] );

Po nadejściu sygnału $state['approval'] zawiera tablicę ładunków sygnałów.

Oczekiwanie na wiele sygnałów

Użyj wait_for_signals(), gdy workflow ma zostać odblokowany po nadejściu dowolnego z kilku sygnałów lub gdy musi zebrać wszystkie wymagane sygnały przed kontynuowaniem.

PHP
use Queuety\Enums\WaitMode;

$workflow_id = Queuety::workflow( 'moderation_gate' )
    ->then( SubmitForReviewHandler::class )
    ->wait_for_signals(
        [ 'editor_approved', 'legal_approved' ],
        WaitMode::All,
        'approvals'
    )
    ->then( PublishContentHandler::class )
    ->dispatch();
  • WaitMode::All czeka, aż zostanie zapisany każdy skonfigurowany sygnał.
  • WaitMode::Any wznawia działanie po nadejściu pierwszego skonfigurowanego sygnału.

W przypadku użycia result_key oczekiwanie na wiele sygnałów przechowuje tablicę asocjacyjną z kluczami będącymi nazwami sygnałów:

PHP
[
    'editor_approved' => [ 'by' => 'editor@example.com' ],
    'legal_approved' => [ 'by' => 'legal@example.com' ],
]

Bez result_key dopasowane ładunki sygnałów są łączone z głównym stanem w kolejności konfiguracji kroku.

Dopasowywanie i korelacja

Oczekiwanie na sygnały może ignorować niepowiązane zdarzenia, dopasowując zawartość ładunku lub korelując zdarzenia na podstawie wspólnego klucza.

PHP
$workflow_id = Queuety::workflow( 'review_gate' )
    ->wait_for_signal(
        name: 'review_ready',
        result_key: 'review',
        match_payload: [ 'source' => 'editorial' ],
        correlation_key: 'task_id',
    )
    ->dispatch( [ 'task_id' => 'task-42' ] );

W tym przykładzie oczekiwanie kończy się tylko wtedy, gdy:

  • nazwa sygnału to review_ready
  • ładunek zawiera source: editorial
  • task_id ładunku pasuje do $state['task_id']

Przydaje się to, gdy kilka systemów zewnętrznych lub agentów publikuje zdarzenia o tych samych nazwach, ale tylko jeden ładunek należy do bieżącego uruchomienia workflow.

Pomocniki zatwierdzania i wprowadzania danych przez człowieka

wait_for_approval() i wait_for_input() to wygodne nakładki na oczekiwanie na sygnały. Działają jak wait_for_signal(), ale domyślnie zapisują wynik pod kluczem z osobną przestrzenią nazw.

PHP
$workflow_id = Queuety::workflow( 'blog_review' )
    ->then( GenerateDraftHandler::class )
    ->wait_for_approval()
    ->wait_for_input()
    ->then( HandleDecisionHandler::class )
    ->dispatch();

Po sygnale approval wartość $state['approval'] zawiera ładunek zatwierdzenia. Po sygnale input wartość $state['input'] zawiera ładunek danych wejściowych.

Użyj wait_for_decision(), gdy chcesz wznowić pojedynczy krok zarówno po zatwierdzeniu, jak i po odrzuceniu:

PHP
$workflow_id = Queuety::workflow( 'legal_review' )
    ->then( DraftContractHandler::class )
    ->wait_for_decision( result_key: 'review' )
    ->then( HandleReviewDecisionHandler::class )
    ->dispatch();

Po nadejściu sygnału $state['review'] zawiera:

PHP
[
    'outcome' => 'approved', // or 'rejected'
    'signal'  => 'approved', // or 'rejected'
    'data'    => [ 'reviewer' => 'ops@example.com' ],
]

Wysyłanie sygnału

Wyślij sygnał do workflow za pomocą Queuety::signal():

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

Gdy sygnał pasuje do oczekiwanego przez workflow, workflow zostaje wznowiony i przechodzi do następnego kroku.

W przypadku procesów z udziałem człowieka fasada udostępnia również metody pomocnicze o znaczeniu semantycznym:

PHP
Queuety::approve_workflow( $workflow_id, [ 'reviewer' => 'editor@example.com' ] );
Queuety::reject_workflow( $workflow_id, [ 'reason' => 'needs citations' ] );
Queuety::submit_workflow_input( $workflow_id, [ 'notes' => 'Ship after copy edit' ] );

Dane sygnału

Możesz przekazać tablicę danych wraz z sygnałem. Dane zostaną połączone ze stanem workflow i będą dostępne w kolejnych krokach:

PHP
Queuety::signal( $workflow_id, 'approved', [
    'approved_by' => 'admin@example.com',
    'approved_at' => time(),
] );

W następnym kroku:

PHP
class PublishContentHandler implements Step {
    public function handle( array $state ): array {
        // $state['approved_by'] is 'admin@example.com'
        // $state['approved_at'] is the timestamp
        // $state['post_id'] is 99
        wp_update_post( [
            'ID'          => $state['post_id'],
            'post_status' => 'publish',
        ] );
        return [ 'published' => true ];
    }

    public function config(): array {
        return [];
    }
}

Klucze zaczynające się od _ (podkreślenia) są zarezerwowane do użytku wewnętrznego i nie są łączone ze stanem.

Wcześniej wysłane sygnały

Sygnały można wysłać, zanim workflow dojdzie do kroku oczekiwania. Po nadejściu sygnał jest zapisywany w tabeli queuety_signals bez względu na bieżący status workflow. Gdy workflow później dojdzie do kroku wait_for_signal, sprawdzi, czy są wcześniejsze sygnały, i od razu będzie kontynuowany, jeśli taki znajdzie.

Oznacza to, że kolejność nie ma znaczenia:

PHP
// Dispatch workflow
$workflow_id = Queuety::workflow( 'order_fulfillment' )
    ->then( PrepareOrderHandler::class )
    ->wait_for_signal( 'payment_confirmed' )
    ->then( ShipOrderHandler::class )
    ->dispatch( [ 'order_id' => 123 ] );

// Signal arrives before the workflow reaches the wait step
Queuety::signal( $workflow_id, 'payment_confirmed', [ 'transaction_id' => 'txn_abc' ] );

// When the workflow finishes PrepareOrderHandler and reaches the wait step,
// it sees the pre-existing signal and continues immediately.

Status workflow podczas oczekiwania na sygnał

Podczas oczekiwania na sygnał status workflow to WorkflowStatus::WaitingForSignal (waiting_for_signal):

PHP
$state = Queuety::workflow_status( $workflow_id );
echo $state->status->value; // 'waiting_for_signal'
echo $state->wait_type;     // 'signal'
echo $state->wait_mode;     // 'all' or 'any'
print_r( $state->waiting_for ); // ['approved']
print_r( $state->wait_details ); // matched/remaining, correlation info, result key, ...

Przykład użycia: zatwierdzenie przez człowieka

Zbuduj workflow zatwierdzania, w którym recenzent musi zatwierdzić treść, zanim zostanie opublikowana:

PHP
$workflow_id = Queuety::workflow( 'blog_review' )
    ->then( GenerateDraftHandler::class )
    ->then( NotifyReviewerHandler::class )
    ->wait_for_decision( result_key: 'review' )
    ->then( HandleDecisionHandler::class )
    ->dispatch( [ 'topic' => 'PHP middleware patterns' ] );

NotifyReviewerHandler wysyła wiadomość e-mail lub na Slack z linkami do zatwierdzenia lub odrzucenia. Linki wywołują kontroler, który wysyła sygnał:

PHP
// In your approval controller
Queuety::approve_workflow(
    $workflow_id,
    [
        'reviewer'    => $current_user->user_email,
        'reviewed_at' => time(),
    ],
    'approved'
);

Krok HandleDecisionHandler odczytuje $state['review']['outcome'] i publikuje wersję roboczą albo przenosi ją do archiwum.

Przykład użycia: wywołanie zwrotne zewnętrznego API

Poczekaj na webhook z usługi zewnętrznej:

PHP
$workflow_id = Queuety::workflow( 'generate_video' )
    ->then( SubmitToRenderServiceHandler::class )
    ->wait_for_signal( 'render_complete' )
    ->then( DownloadVideoHandler::class )
    ->then( NotifyUserHandler::class )
    ->dispatch( [ 'template_id' => 'promo_v2', 'user_id' => 42 ] );

Gdy usługa zewnętrzna wyśle żądanie do punktu końcowego webhooka:

PHP
// In your webhook handler
$payload = json_decode( file_get_contents( 'php://input' ), true );
$workflow_id = (int) $payload['metadata']['workflow_id'];

Queuety::signal( $workflow_id, 'render_complete', [
    'video_url'  => $payload['output_url'],
    'duration'   => $payload['duration_seconds'],
] );

Wiele etapów oczekiwania na sygnały w jednym workflow

Workflow może oczekiwać na sygnały w wielu miejscach:

PHP
Queuety::workflow( 'multi_approval' )
    ->then( CreateProposalHandler::class )
    ->wait_for_signal( 'manager_approval' )
    ->then( EscalateToDirectorHandler::class )
    ->wait_for_signal( 'director_approval' )
    ->then( ExecuteProposalHandler::class )
    ->dispatch( [ 'proposal_id' => 7 ] );

Każdy krok oczekiwania na sygnał jest niezależny. W tym samym workflow możesz łączyć wait_for_signal(), wait_for_signals(), wait_for_approval(), wait_for_input() i wait_for_decision().