Към съдържанието

Сигнали на работния поток

Преглед като Markdown Редактиране в GitHub Документация за 2cc7b2b

Работните потоци могат да се поставят на пауза и да чакат външен вход чрез сигнали. Методите на конструктора wait_for_signal() и wait_for_signals() добавят стъпки, които блокират работния поток, докато приложението ви, оператор или друга система не изпрати очаквания сигнал чрез Queuety::signal().

За потоци с човешка намеса Queuety включва и:

  • wait_for_approval() за пространство от имена за полезния товар на одобрението
  • wait_for_input() за структурирани данни от човек
  • wait_for_decision() за изрични резултати „одобрено“ или „отхвърлено“

За по-голям цялостен пример, който съчетава изпълнение от агенти с контролни точки с човешка намеса, вижте Оркестриране на агенти.

Методът ->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']

Това е полезно, когато няколко външни системи или агенти публикуват едни и същи имена на събития, но само един payload принадлежи на текущото изпълнение на работния процес.

Помощни функции за човешко одобрение и въвеждане на данни

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 с връзки за одобрение или отказ. Тези връзки извикват вашия контролер, който изпраща сигнала:

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

Стъпката HandleDecisionHandler чете $state['review']['outcome'] и публикува или архивира черновата.

Пример: външно обратно извикване чрез API

Изчакайте уебхук от външна услуга:

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

Когато външната услуга изпрати заявка до крайната точка на вашия уебхук:

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() в един и същ работен процес.