Dynamisk for-each

Använd ->for_each() när arbetsflödet inte känner till antalet grenar när det byggs. Queuety läser en array från arbetsflödets tillstånd, skickar ett grenjobb per objekt och går vidare först när det konfigurerade slutförandevillkoret är uppfyllt.

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

Planeringssteget skriver pages till arbetsflödets tillstånd. For-each-steget läser den arrayen och skickar ett grenjobb per objekt.

Grenhanterare

For-each-grenhanterare implementerar 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() tar backoff ett strateginamn: exponential, linear eller fixed. En lista med fördröjningar per försök där ignoreras för närvarande och leder till omedelbara omförsök (#16); ange i stället fördröjningar per försök i stegets definition under backoff.

$state innehåller hela arbetsflödets tillstånd, $item är det enskilda for-each-objektet och $index är dess nollbaserade position.

Slutförandelägen

ForEachMode::All

Steget slutförs först när alla grenar har lyckats. Om någon gren misslyckas permanent misslyckas for-each-steget.

ForEachMode::FirstSuccess

Steget slutförs så snart den första grenen lyckas. Återstående aktiva grenar begravs som ersatta.

ForEachMode::Quorum

Steget slutförs när minst quorum grenar lyckas. Om de återstående grenarna inte längre kan uppfylla kvorumet misslyckas arbetsflödet.

Reducerare

Som standard lagrar Queuety en strukturerad sammanställning under result_key:

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

Om du vill ha en anpassad sammanfattning skickar du med en reducerarklass som implementerar Contracts\ForEachReducer. Sammanställningen lagras fortfarande under result_key, och nycklarna som reduceraren returnerar sammanfogas med arbetsflödets tillstånd:

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

Reducerarens utdata sammanfogas med arbetsflödets tillstånd precis som utdata från andra steg. Den kan också returnera _next_step för att gå vidare till en gren efter att for-each-steget har slutförts.

Omförsöksbeteende

For-each-grenar gör omförsök oberoende av varandra enligt grenhanterarens config(). Om for-each-steget misslyckas och du anropar retry_workflow() expanderar Queuety for-each-steget på nytt och köar bara om grenar som fortfarande saknas eller har misslyckats. Resultat från lyckade grenar finns kvar i arbetsflödets tillstånd.

Välj mellan parallel() och for_each()

Använd parallel() när grenlistan är fast i koden.

Använd for_each() när grenlistan identifieras vid körning utifrån arbetsflödets tillstånd och du behöver slutförandelägen som första framgång eller kvorum.