工作流信号

以 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 回调

等待外部服务发送 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()。