Semnale pentru fluxuri de lucru

Fluxurile de lucru se pot suspenda și pot aștepta date externe folosind semnale. Metodele constructorului wait_for_signal() și wait_for_signals() inserează pași care blochează fluxul de lucru până când aplicația dumneavoastră, un operator sau un alt sistem trimite semnalul așteptat prin Queuety::signal().

Pentru fluxurile cu intervenție umană, Queuety include și:

• wait_for_approval() pentru o structură de date de aprobare cu spațiu de nume
• wait_for_input() pentru date structurate introduse de o persoană
• wait_for_decision() pentru rezultate explicite de aprobare/respingere

Pentru un exemplu mai amplu, de la un capăt la altul, care combină execuțiile agenților cu puncte de control umane, consultați Orchestrarea agenților.

Metoda ->wait_for_signal()

Adăugați un pas de așteptare a unui semnal într-un flux de lucru:

use Queuety\Queuety;

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

Când fluxul de lucru ajunge la pasul wait_for_signal, își setează starea la waiting_for_signal și se suspendă. Nu se execută joburi și nu se consumă resurse în timpul așteptării.

În mod implicit, datele unui singur semnal sunt îmbinate cu starea fluxului de lucru de la nivelul superior. Dacă doriți ca datele să aibă în schimb un spațiu de nume, transmiteți un result_key:

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

După sosirea semnalului, $state['approval'] conține tabloul cu datele semnalului.

Așteptarea mai multor semnale

Folosiți wait_for_signals() când fluxul de lucru trebuie deblocat de oricare dintre mai multe semnale posibile sau când trebuie să colecteze toate semnalele necesare înainte de a continua.

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 așteaptă până când sunt înregistrate toate semnalele configurate.
• WaitMode::Any continuă imediat ce sosește primul semnal configurat.

Cu un result_key, o așteptare pentru mai multe semnale stochează un tablou asociativ indexat după numele semnalului:

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

Fără un result_key, datele semnalelor care corespund sunt îmbinate în starea de nivel superior în ordinea în care a fost configurat pasul.

Potrivire și corelare

Așteptările de semnale pot ignora evenimentele fără legătură, potrivind conținutul datelor sau corelându-le pe baza unei chei comune.

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

În acel exemplu, așteptarea continuă numai când:

• numele semnalului este review_ready
• datele conțin source: editorial
• task_id din date corespunde cu $state['task_id']

Acest lucru este util când mai multe sisteme externe sau agenți publică aceleași nume de evenimente, dar numai datele unei singure sarcini curente de flux de lucru îi aparțin.

Funcții auxiliare pentru aprobarea și introducerea datelor de către oameni

wait_for_approval() și wait_for_input() sunt funcții wrapper convenabile pentru așteptarea semnalelor. Se comportă ca wait_for_signal(), dar folosesc implicit o cheie de rezultat cu spațiu de nume.

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

După semnalul approval, $state['approval'] conține datele aprobării. După semnalul input, $state['input'] conține datele introduse.

Folosiți wait_for_decision() când doriți ca un singur pas să continue fie la aprobare, fie la respingere:

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

După sosirea semnalului, $state['review'] conține:

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

Trimiterea unui semnal

Trimiteți un semnal către un flux de lucru folosind Queuety::signal():

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

Când semnalul corespunde celui așteptat de fluxul de lucru, acesta continuă și trece la pasul următor.

Pentru fluxurile cu intervenție umană, interfața de nivel superior oferă și funcții semantice auxiliare:

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

Datele semnalului

Puteți transmite împreună cu semnalul un tablou de date. Datele sunt îmbinate cu starea fluxului de lucru și devin disponibile pașilor următori:

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

În pasul următor:

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

Cheile care încep cu _ (underscore) sunt rezervate pentru uz intern și nu sunt îmbinate cu starea.

Semnale existente dinainte

Semnalele pot fi trimise înainte ca fluxul de lucru să ajungă la pasul de așteptare. Când sosește un semnal, acesta este înregistrat în tabelul queuety_signals, indiferent de starea curentă a fluxului de lucru. Când fluxul de lucru ajunge ulterior la pasul wait_for_signal, caută semnale existente dinainte și continuă imediat dacă găsește unul.

Asta înseamnă că ordinea nu contează:

// 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.

Starea fluxului de lucru în timpul așteptării unui semnal

În timp ce așteaptă un semnal, starea fluxului de lucru este WorkflowStatus::WaitingForSignal (waiting_for_signal):

$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, ...

Caz de utilizare: aprobare umană

Construiți un flux de lucru de aprobare în care un evaluator uman trebuie să aprobe conținutul înainte de publicare:

$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 trimite un e-mail sau un mesaj Slack cu linkuri de aprobare/respingere. Aceste linkuri apelează controllerul dumneavoastră, care trimite semnalul:

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

Pasul HandleDecisionHandler citește $state['review']['outcome'] și fie publică, fie arhivează ciorna.

Caz de utilizare: callback API extern

Așteptați un webhook de la un serviciu extern:

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

Când serviciul extern trimite o solicitare către endpointul dumneavoastră webhook:

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

Mai multe puncte de așteptare a semnalelor într-un singur flux de lucru

Un flux de lucru poate aștepta semnale în mai multe puncte:

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

Fiecare pas de așteptare a unui semnal este independent. Puteți combina wait_for_signal(), wait_for_signals(), wait_for_approval(), wait_for_input() și wait_for_decision() în același flux de lucru.