Ciclo for-each dinamico
Usate ->for_each() quando il numero di rami del flusso di lavoro non è noto in fase di compilazione. Queuety legge un array dallo stato del flusso di lavoro, avvia un job per ogni elemento e procede solo quando viene soddisfatta la condizione di completamento configurata.
Il metodo ->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 ] );
Il passaggio planner scrive pages nello stato del flusso di lavoro. Il passaggio for-each legge quell'array e avvia un job di ramo per ogni elemento.
Gestori dei rami
I gestori dei rami for-each implementano 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',
];
}
}
In config(), backoff accetta il nome di una strategia: exponential, linear oppure fixed. Un elenco di ritardi per tentativo inserito qui viene attualmente ignorato e causa tentativi immediati (#16); inserite invece i ritardi per tentativo nella definizione del passaggio, in backoff.
$state contiene lo stato completo del flusso di lavoro, $item è il singolo elemento del for-each e $index è la sua posizione a partire da zero.
Modalità di completamento
ForEachMode::All
Il passaggio viene completato solo quando tutti i rami hanno esito positivo. Un errore permanente in qualsiasi ramo causa il fallimento del passaggio for-each.
ForEachMode::FirstSuccess
Il passaggio viene completato non appena il primo ramo ha esito positivo. I rami attivi rimanenti vengono archiviati come superati.
ForEachMode::Quorum
Il passaggio viene completato quando almeno quorum rami hanno esito positivo. Se i rami rimanenti non possono più raggiungere il quorum, il flusso di lavoro non riesce.
Reducer
Per impostazione predefinita, Queuety salva un aggregato strutturato in result_key:
modequorumtotalsucceededfailedsettledwinnerresultsfailures
Per ottenere un riepilogo personalizzato, passate una classe reducer che implementi Contracts\ForEachReducer. L'aggregato viene comunque salvato in result_key e le chiavi restituite dal reducer vengono unite allo stato del flusso di lavoro accanto ad esso:
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,
];
}
}
L'output del reducer viene unito allo stato del flusso di lavoro come qualsiasi altro output di un passaggio. Può anche restituire _next_step per passare a un altro ramo dopo il completamento del for-each.
Comportamento dei tentativi
I rami for-each effettuano tentativi indipendenti in base a config() del gestore del ramo. Se il passaggio for-each non riesce e chiamate retry_workflow(), Queuety espande nuovamente il passaggio e reinserisce in coda solo i rami mancanti o non riusciti. I risultati dei rami riusciti restano nello stato del flusso di lavoro.
Scegliere tra parallel() e for_each()
Usate parallel() quando l'elenco dei rami è definito nel codice.
Usate for_each() quando l'elenco dei rami viene individuato durante l'esecuzione a partire dallo stato del flusso di lavoro e vi servono modalità di completamento come il primo esito positivo o il quorum.