コンテンツへスキップ

ワークフローシグナル

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']と一致する

複数の外部システムやエージェントが同じイベント名を発行するものの、現在のワークフロー実行に属するペイロードが1つだけの場合に便利です。

人による承認と入力のヘルパー

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エンドポイントにPOSTすると、次の処理が行われます。

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

1つのワークフローに複数のシグナルゲートを設定する

ワークフローでは複数の箇所でシグナルを待つことができます。

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()を組み合わせて使えます。