Динамичен цикъл For-Each

Използвайте ->for_each(), когато броят на клоновете в работния поток не е известен по време на изграждането му. Queuety чете масив от състоянието на работния поток, стартира задача за клон за всеки елемент и продължава само когато е изпълнено зададеното условие за приключване.

Методът ->for_each()

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 чете този масив и стартира задача за клон за всеки елемент.

Обработчици на клонове

Обработчиците на клонове за for-each имплементират Contracts\ForEachHandler:

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 е позицията му, броена от нула.

Режими на приключване

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, а върнатите от редуктора ключове се сливат със състоянието на работния поток до него:

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,
        ];
    }
}

Изходът на редуктора се слива със състоянието на работния поток като изхода на всяка друга стъпка. Той може да връща и _next_step за разклоняване след приключването на for-each.

Повторни опити

Клоновете на for-each правят повторни опити независимо според config() на обработчика на клона. Ако стъпката for-each се провали и извикате retry_workflow(), Queuety отново разширява стъпката for-each и поставя в опашката само липсващите или неуспешните клонове. Резултатите от успешните клонове остават в състоянието на работния поток.

Избор между parallel() и for_each()

Използвайте parallel(), когато списъкът с клонове е зададен в кода.

Използвайте for_each(), когато списъкът с клонове се открива по време на изпълнение от състоянието на работния поток и са необходими режими на приключване като първи успешен клон или кворум.