For-Each dinamic

Folosiți ->for_each() când numărul de ramuri din fluxul de lucru nu este cunoscut în momentul construirii. Queuety citește un tablou din starea fluxului de lucru, expediază o sarcină de ramură pentru fiecare element și avansează numai când este îndeplinită condiția de finalizare configurată.

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

Pasul planificatorului scrie pages în starea fluxului de lucru. Pasul for-each citește tabloul și expediază o sarcină de ramură pentru fiecare element.

Gestionarele ramurilor

Gestionarele ramurilor for-each implementează 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',
        ];
    }
}

În config(), backoff primește un nume de strategie: exponential, linear sau fixed. O listă de întârzieri per încercare este în prezent eliminată și devine o serie de reîncercări imediate (#16); plasați întârzierile per încercare în backoff din definiția pasului.

$state conține întreaga stare a fluxului de lucru, $item este elementul for-each individual, iar $index este poziția sa indexată de la zero.

Moduri de finalizare

ForEachMode::All

Pasul se încheie numai după ce toate ramurile reușesc. Orice eșec permanent al unei ramuri provoacă eșecul pasului for-each.

ForEachMode::FirstSuccess

Pasul se încheie de îndată ce prima ramură reușește. Ramurile active rămase sunt îngropate ca fiind înlocuite.

ForEachMode::Quorum

Pasul se încheie când reușesc cel puțin quorum ramuri. Dacă ramurile rămase nu mai pot îndeplini cvorumul, fluxul de lucru eșuează.

Reducătoare

În mod implicit, Queuety stochează un agregat structurat în result_key:

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

Dacă doriți un rezumat personalizat, transmiteți o clasă de reducător care implementează Contracts\ForEachReducer. Agregatul este stocat în continuare în result_key, iar cheile returnate de reducător sunt îmbinate în starea fluxului de lucru alături de acesta:

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

Rezultatul reducătorului este îmbinat în starea fluxului de lucru la fel ca rezultatul oricărui alt pas. Reducătorul poate returna și _next_step pentru a ramifica fluxul după încheierea pasului for-each.

Comportamentul reîncercărilor

Ramurile for-each se reîncearcă independent, conform config() al gestionarului ramurii. Dacă pasul for-each eșuează și apelați retry_workflow(), Queuety extinde din nou pasul for-each și repune în coadă numai ramurile care lipsesc sau au eșuat. Rezultatele ramurilor reușite rămân în starea fluxului de lucru.

Alegerea între parallel() și for_each()

Folosiți parallel() când lista ramurilor este fixată în cod.

Folosiți for_each() când lista ramurilor este descoperită în timpul execuției din starea fluxului de lucru și aveți nevoie de moduri de finalizare precum prima reușită sau cvorumul.