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

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

Для процессов с участием человека Queuety также включает:

• wait_for_approval() для пространства имен полезной нагрузки подтверждения
• wait_for_input() для структурированного ввода от человека
• wait_for_decision() для явных результатов одобрения или отклонения

Более подробный пример от начала до конца, объединяющий запуски агентов с проверками человеком, см. в разделе Оркестрация агентов.

Метод ->wait_for_signal()

Добавьте в рабочий процесс шаг ожидания сигнала:

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:

$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(), если рабочий процесс должен продолжиться при поступлении одного из нескольких возможных сигналов или должен собрать все обязательные сигналы, прежде чем продолжить.

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 ожидание нескольких сигналов сохраняет ассоциативный массив с ключами по именам сигналов:

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

Без result_key полезные нагрузки подходящих сигналов объединяются с состоянием верхнего уровня в порядке настройки шага.

Сопоставление и корреляция

Ожидание сигнала может игнорировать посторонние события, сопоставляя содержимое полезной нагрузки или используя общий ключ для корреляции.

$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(), но по умолчанию используют ключ результата в пространстве имен.

$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(), если нужно, чтобы один шаг продолжал выполнение при одобрении или отклонении:

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

После поступления сигнала $state['review'] содержит:

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

Отправка сигнала

Отправьте сигнал рабочему процессу с помощью Queuety::signal():

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

Когда сигнал совпадает с ожидаемым рабочим процессом, процесс продолжается и переходит к следующему шагу.

Для сценариев с участием человека фасад также предоставляет семантические вспомогательные методы:

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

Данные сигнала

Вместе с сигналом можно передать массив данных. Эти данные объединяются с состоянием рабочего процесса и становятся доступны последующим шагам:

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

На следующем шаге:

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, он проверяет ранее поступившие сигналы и сразу продолжает выполнение, если находит подходящий.

Это означает, что порядок не имеет значения:

// 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):

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

Сценарий использования: подтверждение человеком

Создайте процесс подтверждения, в котором проверяющий должен одобрить содержимое до его публикации:

$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 со ссылками для одобрения или отклонения. Эти ссылки вызывают ваш контроллер, который отправляет сигнал:

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

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

Сценарий использования: внешний обратный вызов API

Ожидайте вебхук от внешней службы:

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

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

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

Несколько точек ожидания сигналов в одном рабочем процессе

Рабочий процесс может ожидать сигналы в нескольких точках:

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