動的for-each

Markdownとして表示 GitHubで編集ドキュメント: 2cc7b2b

ビルド時に分岐数が決まらないワークフローでは->for_each()を使います。Queuetyはワークフローの状態から配列を読み取り、項目ごとに分岐ジョブを1つディスパッチし、設定された完了条件が満たされた場合にのみ次へ進みます。

`->for_each()`メソッド

PHP

use Queuety\Contracts\ForEachHandler;
use Queuety\Contracts\ForEachReducer;
use Queuety\Enums\ForEachMode;
use Queuety\Queuety;

Queuety::workflow( 'index_site' )
    ->then( PlanPagesStep::class )
    ->for_each(
        items_key: 'pages',
        handler_class: IndexPageStep::class,
        result_key: 'page_results',
        mode: ForEachMode::Quorum,
        quorum: 2,
        reducer_class: IndexSummaryReducer::class,
        name: 'index_pages',
    )
    ->then( PublishIndexStep::class )
    ->dispatch( [ 'site_id' => 42 ] );

プランナーステップはpagesをワークフローの状態に書き込みます。for-eachステップはその配列を読み取り、項目ごとに分岐ジョブを1つディスパッチします。

分岐ハンドラー

for-eachの分岐ハンドラーはContracts\ForEachHandlerを実装します:

PHP

use Queuety\Contracts\ForEachHandler;

final class IndexPageStep implements ForEachHandler {
    public function handle_item( array $state, mixed $item, int $index ): array {
        $page = is_array( $item ) ? $item : [ 'url' => (string) $item ];

        return [
            'url' => $page['url'],
            'title' => fetch_title( $page['url'] ),
            'index' => $index,
        ];
    }

    public function config(): array {
        return [
            'max_attempts' => 3,
            'backoff' => 'exponential',
        ];
    }
}

config()では、backoffに戦略名としてexponential、linear、fixedのいずれかを指定します。ここで試行ごとの遅延リストを指定しても、現在は破棄されて即時再試行になるため（#16）、試行ごとの遅延はステップ定義のbackoffに指定してください。

$stateにはワークフローの全状態が入り、$itemには個々のfor-each項目が入り、$indexにはその0始まりの位置が入ります。

完了モード

ForEachMode::All

すべての分岐が成功した後にのみ、ステップが完了します。分岐が恒久的に失敗すると、for-eachステップも失敗します。

ForEachMode::FirstSuccess

最初の分岐が成功した時点でステップが完了します。残りの実行中の分岐は、優先対象外として処理されます。

ForEachMode::Quorum

quorum個以上の分岐が成功した時点でステップが完了します。残りの分岐では必要な定足数を満たせなくなった場合、ワークフローは失敗します。

リデューサー

デフォルトでは、Queuetyは構造化された集約結果をresult_keyの下に保存します:

mode
quorum
total
succeeded
failed
settled
winner
results
failures

独自の要約を作成するには、Contracts\ForEachReducerを実装したリデューサークラスを渡します。集約結果は引き続きresult_keyに保存され、リデューサーが返すキーはその隣にワークフロー状態へマージされます。

PHP

use Queuety\Contracts\ForEachReducer;

final class IndexSummaryReducer implements ForEachReducer {
    public function reduce( array $state, array $for_each ): array {
        return [
            'indexed_count' => count( $for_each['results'] ),
            'winning_url' => $for_each['winner']['output']['url'] ?? null,
        ];
    }
}

リデューサーの出力は、他のステップ出力と同様にワークフロー状態へマージされます。また、for-eachの完了後に分岐するため、_next_stepを返すこともできます。

再試行の動作

for-eachの各ブランチは、ブランチハンドラーのconfig()に従って個別に再試行されます。for-eachステップが失敗し、retry_workflow()を呼び出すと、Queuetyはfor-eachステップを再展開し、未完了または失敗したブランチのみを再キューに追加します。成功したブランチの結果はワークフロー状態に残ります。

`parallel()`と`for_each()`の使い分け

ブランチのリストがコード内で固定されている場合は、parallel()を使います。

ブランチのリストが実行時にワークフロー状態から検出され、first-successやquorumなどの完了モードが必要な場合は、for_each()を使います。