Vai al contenuto

Segnali del flusso di lavoro

Visualizza come Markdown Modifica su GitHub Documentazione di 2cc7b2b

I flussi di lavoro possono essere sospesi in attesa di input esterni tramite i segnali. I metodi builder wait_for_signal() e wait_for_signals() inseriscono passaggi che bloccano il flusso di lavoro finché l'applicazione, un operatore o un altro sistema non invia il segnale atteso tramite Queuety::signal().

Per i flussi con intervento umano, Queuety include anche:

  • wait_for_approval() per un payload di approvazione con namespace
  • wait_for_input() per dati umani strutturati
  • wait_for_decision() per esiti espliciti di approvazione o rifiuto

Per un esempio più ampio dall'inizio alla fine che combina esecuzioni di agenti e passaggi di approvazione umana, consultate Orchestrazione di agenti.

Il metodo ->wait_for_signal()

Aggiungere un passaggio di attesa di un segnale a un flusso di lavoro:

PHP
use Queuety\Queuety;

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

Quando il flusso di lavoro raggiunge il passaggio wait_for_signal, imposta il proprio stato su waiting_for_signal e si sospende. Durante l'attesa non ci sono job in esecuzione né risorse consumate.

Per impostazione predefinita, il payload di un singolo segnale viene unito allo stato del flusso di lavoro di primo livello. Per assegnare invece un namespace al payload, passate un result_key:

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

Dopo l'arrivo del segnale, $state['approval'] contiene l'array dei payload dei segnali.

Attendere più segnali

Usate wait_for_signals() quando il flusso di lavoro deve sbloccarsi alla ricezione di uno tra vari segnali possibili oppure deve raccogliere tutti i segnali richiesti prima di proseguire.

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 attende finché non sono stati registrati tutti i segnali configurati.
  • WaitMode::Any riprende non appena arriva il primo segnale configurato.

Con un result_key, un'attesa di più segnali salva un array associativo indicizzato per nome del segnale:

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

Senza un result_key, i payload dei segnali corrispondenti vengono uniti allo stato di primo livello nell'ordine in cui è stato configurato il passaggio.

Corrispondenza e correlazione

Le attese dei segnali possono ignorare gli eventi non pertinenti confrontando il contenuto del payload o usando una chiave condivisa per la correlazione.

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

In questo esempio, l'attesa riprende solo quando:

  • il nome del segnale è review_ready
  • il payload contiene source: editorial
  • il valore task_id del payload corrisponde a $state['task_id']

Questo è utile quando più sistemi esterni o agenti pubblicano gli stessi nomi di evento, ma un solo payload appartiene all'esecuzione corrente del flusso di lavoro.

Helper per l'approvazione e gli input umani

wait_for_approval() e wait_for_input() sono wrapper pratici per le attese dei segnali. Si comportano come wait_for_signal(), ma usano per impostazione predefinita una chiave del risultato con namespace.

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

Dopo il segnale approval, $state['approval'] contiene il payload di approvazione. Dopo il segnale input, $state['input'] contiene il payload di input.

Usate wait_for_decision() quando volete che un singolo passaggio riprenda sia in caso di approvazione sia in caso di rifiuto:

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

Dopo l'arrivo del segnale, $state['review'] contiene:

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

Inviare un segnale

Inviare un segnale a un flusso di lavoro usando Queuety::signal():

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

Quando il segnale corrisponde a quello atteso dal flusso di lavoro, questo riprende e avanza al passaggio successivo.

Per i flussi con intervento umano, la facade espone anche helper semantici:

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

Dati del segnale

Potete passare un array di dati insieme al segnale. I dati vengono uniti allo stato del flusso di lavoro e resi disponibili ai passaggi successivi:

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

Nel passaggio successivo:

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

Le chiavi che iniziano con _ (underscore) sono riservate all'uso interno e non vengono unite allo stato.

Segnali preesistenti

È possibile inviare segnali prima che il flusso di lavoro raggiunga il passaggio di attesa. Quando arriva un segnale, viene registrato nella tabella queuety_signals indipendentemente dallo stato attuale del flusso di lavoro. Quando in seguito il flusso di lavoro raggiunge il passaggio wait_for_signal, controlla se esistono segnali precedenti e prosegue subito se ne trova uno.

Questo significa che l'ordine non è importante:

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.

Stato del flusso di lavoro durante l'attesa di un segnale

In attesa di un segnale, lo stato del flusso di lavoro è 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, ...

Caso d'uso: approvazione umana

Create un flusso di approvazione in cui un revisore umano deve approvare i contenuti prima che vengano pubblicati:

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

Il NotifyReviewerHandler invia un'e-mail o un messaggio Slack con link per approvare o rifiutare. Questi link chiamano il controller, che invia il segnale:

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

Il passaggio HandleDecisionHandler legge $state['review']['outcome'] e pubblica o archivia la bozza.

Caso d'uso: callback API esterno

Attendete un webhook da un servizio esterno:

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

Quando il servizio esterno invia una richiesta all'endpoint del webhook:

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

Più punti di attesa per i segnali in un flusso di lavoro

Un flusso di lavoro può attendere segnali in più punti:

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

Ogni passaggio di attesa del segnale è indipendente. Potete combinare wait_for_signal(), wait_for_signals(), wait_for_approval(), wait_for_input() e wait_for_decision() nello stesso flusso di lavoro.