Dynamiczne For-Each

Wyświetl jako Markdown Edytuj w GitHub Dokumentacja: 2cc7b2b

Użyj ->for_each(), gdy liczba gałęzi przepływu pracy nie jest znana podczas jego tworzenia. Queuety odczytuje tablicę ze stanu przepływu pracy, uruchamia osobne zadanie gałęzi dla każdego elementu i przechodzi dalej dopiero po spełnieniu skonfigurowanego warunku ukończenia.

Metoda `->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 ] );

Krok planisty zapisuje pages w stanie przepływu pracy. Krok for-each odczytuje tę tablicę i uruchamia osobne zadanie gałęzi dla każdego elementu.

Obsługa gałęzi

Procedury obsługi gałęzi for-each implementują 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',
        ];
    }
}

W config() backoff przyjmuje nazwę strategii: exponential, linear lub fixed. Lista opóźnień dla kolejnych prób podana w tym miejscu jest obecnie odrzucana, co skutkuje natychmiastowymi ponownymi próbami (#16); opóźnienia dla kolejnych prób należy zamiast tego ustawić w definicji kroku w backoff.

$state zawiera pełny stan przepływu pracy, $item to pojedynczy element for-each, a $index to jego pozycja liczona od zera.

Tryby ukończenia

ForEachMode::All

Krok kończy się dopiero po pomyślnym zakończeniu wszystkich gałęzi. Każda trwała awaria gałęzi powoduje niepowodzenie kroku for-each.

ForEachMode::FirstSuccess

Krok kończy się, gdy tylko pierwsza gałąź zakończy się pomyślnie. Pozostałe aktywne gałęzie zostają ukryte jako zastąpione.

ForEachMode::Quorum

Krok kończy się po pomyślnym zakończeniu co najmniej quorum gałęzi. Jeśli pozostałe gałęzie nie mogą już spełnić kworum, przepływ pracy kończy się niepowodzeniem.

Reduktory

Domyślnie Queuety zapisuje ustrukturyzowaną agregację pod kluczem result_key:

mode
quorum
total
succeeded
failed
settled
winner
results
failures

Aby utworzyć własne podsumowanie, przekaż klasę reduktora implementującą Contracts\ForEachReducer. Agregacja nadal jest zapisywana pod kluczem result_key, a zwrócone przez reduktor klucze są scalane z sąsiadującymi z nim danymi stanu przepływu pracy:

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

Wynik reduktora jest scalany ze stanem przepływu pracy tak samo jak wynik dowolnego innego kroku. Reduktor może też zwrócić _next_step, aby po zakończeniu for-each uruchomić kolejną gałąź.

Ponawianie prób

Próby w gałęziach for-each są ponawiane niezależnie, zgodnie z ustawieniem config() procedury obsługi gałęzi. Jeśli krok for-each zakończy się niepowodzeniem i wywołasz retry_workflow(), Queuety ponownie rozwinie krok for-each i umieści w kolejce tylko brakujące lub zakończone niepowodzeniem gałęzie. Wyniki pomyślnych gałęzi pozostają w stanie przepływu pracy.

Wybór między `parallel()` a `for_each()`

Użyj parallel(), gdy lista gałęzi jest ustalona w kodzie.

Użyj for_each(), gdy lista gałęzi jest ustalana w czasie działania na podstawie stanu przepływu pracy i potrzebujesz trybów ukończenia, takich jak pierwsza pomyślna gałąź lub kworum.