Перейти до вмісту

Сигнали робочого процесу

Переглянути як 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']

Це корисно, коли кілька зовнішніх систем або агентів надсилають події з однаковими назвами, але лише одні дані стосуються поточного запуску робочого процесу.

Допоміжні засоби для підтвердження та введення даних людиною

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().