Dynamisk for-each

Brug ->for_each(), når workflowet ikke kender antallet af grene, når det bygges. Queuety læser et array fra workflowtilstanden, sender ét grenjob pr. element og fortsætter først, når den konfigurerede fuldførelsesbetingelse er opfyldt.

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

Plannertrinnet skriver pages til workflowtilstanden. For-each-trinnet læser arrayet og sender ét grenjob pr. element.

Grenhåndterere

For-each-grenhåndterere implementerer 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',
        ];
    }
}

I config() tager backoff et strateginavn: exponential, linear eller fixed. En liste over forsinkelser pr. forsøg droppes i øjeblikket og bliver til øjeblikkelige nye forsøg (#16). Angiv i stedet forsinkelser pr. forsøg i trindefinitionens backoff.

$state indeholder hele workflowtilstanden, $item er det enkelte for-each-element, og $index er dets nulbaserede position.

Fuldførelsestilstande

ForEachMode::All

Trinnet afsluttes først, når alle grene lykkes. En permanent fejl i en gren får for-each-trinnet til at fejle.

ForEachMode::FirstSuccess

Trinnet afsluttes, så snart den første gren lykkes. De resterende aktive grene markeres som overhalet.

ForEachMode::Quorum

Trinnet afsluttes, når mindst quorum grene lykkes. Hvis de resterende grene ikke længere kan opfylde quorumkravet, fejler workflowet.

Reduktionsfunktioner

Som standard gemmer Queuety en struktureret sammenfatning under result_key:

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

Hvis du vil have en tilpasset sammenfatning, skal du angive en reduktionsklasse, der implementerer Contracts\ForEachReducer. Sammenfatningen gemmes stadig under result_key, og de nøgler, som reduktionsfunktionen returnerer, flettes ind i workflowtilstanden ved siden af den:

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

Output fra reduktionsfunktionen flettes ind i workflowtilstanden som output fra ethvert andet trin. Den kan også returnere _next_step til forgrening, efter at for-each-trinnet er afsluttet.

Adfærd ved nye forsøg

For-each-grene prøves igen uafhængigt efter grenens config(). Hvis for-each-trinnet fejler, og du kalder retry_workflow(), udvider Queuety for-each-trinnet igen og sætter kun manglende eller fejlede grene i kø igen. Resultater fra vellykkede grene forbliver i workflowtilstanden.

Valg mellem parallel() og for_each()

Brug parallel(), når grenlisten er fastlagt i koden.

Brug for_each(), når grenlisten findes under kørsel fra workflowtilstanden, og du har brug for fuldførelsestilstande som første succes eller quorum.