Dinamični for-each

Uporabite ->for_each(), ko število vej delovnega toka ob gradnji še ni znano. Queuety prebere tabelo iz stanja delovnega toka, odpremi eno opravilo veje za vsak element in nadaljuje šele, ko je izpolnjen nastavljen pogoj dokončanja.

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

Korak načrtovalnika zapiše pages v stanje delovnega toka. Korak for-each prebere to tabelo in odpremi eno opravilo veje za vsak element.

Obdelovalniki vej

Obdelovalniki vej for-each izvajajo 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',
        ];
    }
}

V config() backoff prejme ime strategije: exponential, linear ali fixed. Seznam zamikov med posameznimi poskusi se trenutno tam zavrže in povzroči takojšnje ponovne poskuse (#16); zamike za posamezne poskuse namesto tega določite v definiciji koraka, v backoff.

$state vsebuje celotno stanje delovnega toka, $item je trenutni element for-each, $index pa njegov položaj, ki se začne z ničlo.

Načini dokončanja

ForEachMode::All

Korak se zaključi šele, ko uspejo vse veje. Trajna napaka katere koli veje povzroči neuspeh koraka for-each.

ForEachMode::FirstSuccess

Korak se zaključi takoj, ko uspe prva veja. Preostale aktivne veje so označene kot nadomeščene.

ForEachMode::Quorum

Korak se zaključi, ko uspe vsaj quorum vej. Če preostale veje ne morejo več doseči zahtevanega kvoruma, delovni tok ne uspe.

Reducerji

Queuety privzeto shrani strukturiran zbirni rezultat pod result_key:

• mode
• quorum
• total
• succeeded
• failed
• settled
• winner
• results
• failures

Če želite povzetek po meri, podajte razred reducerja, ki izvaja Contracts\ForEachReducer. Zbirni rezultat je še vedno shranjen pod result_key, ključi, ki jih vrne reducer, pa se združijo s stanjem delovnega toka:

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

Izhod reducerja se združi s stanjem delovnega toka enako kot izhod katerega koli drugega koraka. Vrne lahko tudi _next_step za prehod v vejo po zaključku for-each.

Vedenje pri ponovnih poskusih

Veje for-each neodvisno ponavljajo poskuse v skladu z config() svojega obdelovalnika. Če korak for-each ne uspe in pokličete retry_workflow(), Queuety znova razširi korak for-each in ponovno uvrsti v čakalno vrsto samo manjkajoče ali neuspešne veje. Rezultati uspešnih vej ostanejo v stanju delovnega toka.

Izbira med parallel() in for_each()

Uporabite parallel(), kadar je seznam vej določen v kodi.

Uporabite for_each(), kadar se seznam vej med izvajanjem odkrije v stanju delovnega toka in potrebujete načine dokončanja, kot sta prvi uspeh ali kvorum.