Μετάβαση στο περιεχόμενο

Σήματα ροών εργασιών

Προβολή ως Markdown Επεξεργασία στο GitHub Τεκμηρίωση στο 2cc7b2b

Οι ροές εργασιών μπορούν να τεθούν σε παύση και να περιμένουν εξωτερική είσοδο μέσω σημάτων. Οι μέθοδοι κατασκευαστή wait_for_signal() και wait_for_signals() εισάγουν βήματα που εμποδίζουν τη συνέχιση της ροής μέχρι η εφαρμογή σας, ένας χειριστής ή κάποιο άλλο σύστημα να στείλει το αναμενόμενο σήμα μέσω Queuety::signal().

Για ροές με συμμετοχή ανθρώπου, το Queuety περιλαμβάνει επίσης:

  • wait_for_approval() για φορτίο έγκρισης με πεδίο ονομάτων
  • wait_for_input() για δομημένη ανθρώπινη είσοδο
  • wait_for_decision() για ρητά αποτελέσματα έγκρισης ή απόρριψης

Για ένα εκτενέστερο παράδειγμα από άκρη σε άκρη που συνδυάζει εκτελέσεις agent με ανθρώπινες πύλες ελέγχου, δείτε την ενότητα Agent Orchestration.

Η μέθοδος ->wait_for_signal()

Προσθέστε ένα βήμα αναμονής σήματος σε μια ροή εργασιών:

PHP
use Queuety\Queuety;

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

Όταν η ροή εργασιών φτάσει στο βήμα wait_for_signal, η κατάστασή της ορίζεται σε waiting_for_signal και τίθεται σε παύση. Δεν εκτελούνται εργασίες ούτε καταναλώνονται πόροι όσο περιμένει.

Από προεπιλογή, το φορτίο ενός σήματος συγχωνεύεται με την κατάσταση ανώτατου επιπέδου της ροής εργασιών. Αν θέλετε το φορτίο να έχει δικό του πεδίο ονομάτων, περάστε ένα result_key:

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

Μετά την άφιξη του σήματος, το $state['approval'] περιέχει τον πίνακα φορτίων σημάτων.

Αναμονή για πολλά σήματα

Χρησιμοποιήστε το wait_for_signals() όταν η ροή εργασιών πρέπει να ξεμπλοκάρει με οποιοδήποτε από πολλά πιθανά σήματα ή όταν πρέπει να συγκεντρώσει όλα τα απαιτούμενα σήματα πριν συνεχίσει.

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 περιμένει μέχρι να καταγραφεί κάθε διαμορφωμένο σήμα.
  • Το WaitMode::Any συνεχίζει μόλις φτάσει το πρώτο διαμορφωμένο σήμα.

Με ένα result_key, η αναμονή πολλών σημάτων αποθηκεύει έναν συσχετιστικό πίνακα με κλειδί το όνομα του σήματος:

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

Χωρίς result_key, τα φορτία των σημάτων που ταιριάζουν συγχωνεύονται με την κατάσταση ανώτατου επιπέδου, με τη σειρά που διαμορφώθηκε το βήμα.

Αντιστοίχιση και συσχέτιση

Οι αναμονές σημάτων μπορούν να αγνοούν άσχετα συμβάντα αντιστοιχίζοντας το περιεχόμενο του φορτίου ή συσχετίζοντας τα σήματα μέσω ενός κοινού κλειδιού.

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

Σε αυτό το παράδειγμα, η αναμονή συνεχίζεται μόνο όταν:

  • το όνομα του σήματος είναι review_ready
  • το φορτίο περιέχει source: editorial
  • το task_id του φορτίου αντιστοιχεί στο $state['task_id']

Αυτό είναι χρήσιμο όταν πολλά εξωτερικά συστήματα ή agent δημοσιεύουν τα ίδια ονόματα συμβάντων, αλλά μόνο ένα φορτίο ανήκει στην τρέχουσα εκτέλεση της ροής εργασιών.

Βοηθητικές λειτουργίες για ανθρώπινη έγκριση και είσοδο

Τα wait_for_approval() και wait_for_input() είναι βοηθητικές λειτουργίες γύρω από τις αναμονές σημάτων. Συμπεριφέρονται όπως το wait_for_signal(), αλλά χρησιμοποιούν από προεπιλογή ένα κλειδί αποτελέσματος με δικό του πεδίο ονομάτων.

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

Μετά το σήμα approval, το $state['approval'] περιέχει το φορτίο έγκρισης. Μετά το σήμα input, το $state['input'] περιέχει το φορτίο εισόδου.

Χρησιμοποιήστε το wait_for_decision() όταν θέλετε ένα μόνο βήμα να συνεχίζεται είτε μετά από έγκριση είτε μετά από απόρριψη:

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

Μετά την άφιξη του σήματος, το $state['review'] περιέχει:

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

Αποστολή σήματος

Στείλτε ένα σήμα σε μια ροή εργασιών χρησιμοποιώντας το Queuety::signal():

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

Όταν το σήμα αντιστοιχεί σε αυτό που περιμένει η ροή εργασιών, η ροή συνεχίζεται και προχωρά στο επόμενο βήμα.

Για ροές με συμμετοχή ανθρώπου, η πρόσοψη παρέχει επίσης σημασιολογικές βοηθητικές λειτουργίες:

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

Δεδομένα σήματος

Μπορείτε να περάσετε έναν πίνακα δεδομένων μαζί με το σήμα. Τα δεδομένα συγχωνεύονται με την κατάσταση της ροής εργασιών και γίνονται διαθέσιμα στα επόμενα βήματα:

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

Στο επόμενο βήμα:

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

Τα κλειδιά που αρχίζουν με _ (κάτω παύλα) προορίζονται για εσωτερική χρήση και δεν συγχωνεύονται με την κατάσταση.

Προϋπάρχοντα σήματα

Τα σήματα μπορούν να σταλούν πριν η ροή εργασιών φτάσει στο βήμα αναμονής. Όταν φτάσει ένα σήμα, καταγράφεται στον πίνακα queuety_signals ανεξάρτητα από την τρέχουσα κατάσταση της ροής εργασιών. Όταν η ροή φτάσει αργότερα στο βήμα wait_for_signal, ελέγχει για προϋπάρχοντα σήματα και συνεχίζει αμέσως αν βρει κάποιο.

Αυτό σημαίνει ότι η σειρά δεν έχει σημασία:

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.

Κατάσταση ροής εργασιών κατά την αναμονή σήματος

Κατά την αναμονή σήματος, η κατάσταση της ροής εργασιών είναι 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, ...

Περίπτωση χρήσης: ανθρώπινη έγκριση

Δημιουργήστε μια ροή έγκρισης στην οποία ένας ανθρώπινος αξιολογητής πρέπει να εγκρίνει το περιεχόμενο πριν δημοσιευτεί:

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 στέλνει ένα μήνυμα ηλεκτρονικού ταχυδρομείου ή ένα μήνυμα Slack με συνδέσμους έγκρισης ή απόρριψης. Οι σύνδεσμοι καλούν τον controller σας, ο οποίος στέλνει το σήμα:

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

Το βήμα HandleDecisionHandler διαβάζει το $state['review']['outcome'] και είτε δημοσιεύει είτε αρχειοθετεί το προσχέδιο.

Περίπτωση χρήσης: εξωτερική επανάκληση API

Περιμένετε ένα webhook από εξωτερική υπηρεσία:

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

Όταν η εξωτερική υπηρεσία στείλει αίτημα στο τελικό σημείο 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'],
] );

Πολλαπλές πύλες σημάτων σε μία ροή εργασιών

Μια ροή εργασιών μπορεί να περιμένει σήματα σε πολλά σημεία:

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

Κάθε βήμα αναμονής σήματος είναι ανεξάρτητο. Μπορείτε να συνδυάσετε τα wait_for_signal(), wait_for_signals(), wait_for_approval(), wait_for_input() και wait_for_decision() στην ίδια ροή εργασιών.