Dynamische For-Each

Bekijken als Markdown Bewerken op GitHub Documentatie op 2cc7b2b

Gebruik ->for_each() wanneer de workflow het aantal vertakkingen niet kent tijdens het bouwen. Queuety leest een array uit de workflowstatus, verzendt één taak per item voor een vertakking en gaat pas verder wanneer aan de ingestelde voltooiingsvoorwaarde is voldaan.

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

De plannerstap schrijft pages naar de workflowstatus. De for-each-stap leest die array en verzendt één taak per item voor een vertakking.

Vertakkingshandlers

Handlers voor for-each-vertakkingen implementeren 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',
        ];
    }
}

In config() neemt backoff een strategienaam: exponential, linear of fixed. Een lijst met vertragingen per poging wordt daar momenteel genegeerd en leidt tot onmiddellijke nieuwe pogingen (#16); stel vertragingen per poging in de backoff van de stapdefinitie in.

$state bevat de volledige workflowstatus, $item is het afzonderlijke for-each-item en $index is de positie ervan vanaf nul.

Voltooiingsmodi

ForEachMode::All

De stap wordt pas afgerond wanneer elke vertakking slaagt. Als een vertakking definitief mislukt, mislukt de for-each-stap.

ForEachMode::FirstSuccess

De stap wordt afgerond zodra de eerste vertakking slaagt. De overige actieve vertakkingen worden als vervangen gemarkeerd.

ForEachMode::Quorum

De stap wordt afgerond zodra ten minste quorum vertakkingen slagen. Als de resterende vertakkingen het quorum niet meer kunnen halen, mislukt de workflow.

Reducers

Standaard slaat Queuety een gestructureerde aggregatie op onder result_key:

mode
quorum
total
succeeded
failed
settled
winner
results
failures

Als je een aangepaste samenvatting wilt, geef dan een reducerklasse door die Contracts\ForEachReducer implementeert. De aggregatie wordt nog steeds opgeslagen onder result_key, en de sleutels die de reducer retourneert, worden ernaast samengevoegd met de workflowstatus:

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

De uitvoer van de reducer wordt net als elke andere stapuitvoer samengevoegd met de workflowstatus. De reducer kan ook _next_step retourneren om na voltooiing van de for-each verder te vertakken.

Gedrag bij nieuwe pogingen

For-each-vertakkingen doen onafhankelijk nieuwe pogingen volgens config() van de vertakkingshandler. Als de for-each-stap mislukt en je retry_workflow() aanroept, vouwt Queuety de for-each-stap opnieuw uit en plaatst alleen vertakkingen die nog ontbreken of mislukt zijn opnieuw in de wachtrij. Geslaagde vertakkingsresultaten blijven in de workflowstatus staan.

Kiezen tussen `parallel()` en `for_each()`

Gebruik parallel() als de vertakkingslijst vastligt in de code.

Gebruik for_each() als de vertakkingslijst tijdens runtime wordt ontdekt op basis van de workflowstatus en je voltooiingsmodi zoals eerste succes of quorum nodig hebt.