Bucle dinámico para cada elemento
Usa ->for_each() cuando el flujo de trabajo no conoce la cantidad de ramas en el momento de la compilación. Queuety lee un arreglo del estado del flujo de trabajo, envía un trabajo de rama por elemento y avanza solo cuando se cumple la condición de finalización configurada.
El método ->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 ] );
El paso de planificación escribe pages en el estado del flujo de trabajo. El paso de bucle para cada elemento lee ese arreglo y envía un trabajo de rama por elemento.
Controladores de ramas
Los controladores de ramas del bucle para cada elemento implementan 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',
];
}
}
En config(), backoff recibe un nombre de estrategia: exponential, linear o fixed. Actualmente, allí se descarta una lista de demoras por intento y se convierte en reintentos inmediatos (#16); define las demoras por intento en backoff de la definición del paso.
$state contiene el estado completo del flujo de trabajo, $item es el elemento individual del bucle y $index es su posición, empezando por cero.
Modos de finalización
ForEachMode::All
El paso solo se completa cuando todas las ramas tienen éxito. Cualquier fallo permanente de una rama hace que falle el bucle para cada elemento.
ForEachMode::FirstSuccess
El paso se completa en cuanto tiene éxito la primera rama. Las ramas activas restantes se archivan como reemplazadas.
ForEachMode::Quorum
El paso se completa cuando tienen éxito al menos quorum ramas. Si las ramas restantes ya no pueden alcanzar el quórum, el flujo de trabajo falla.
Reductores
De forma predeterminada, Queuety almacena un agregado estructurado en result_key:
modequorumtotalsucceededfailedsettledwinnerresultsfailures
Si quieres un resumen personalizado, pasa una clase reductora que implemente Contracts\ForEachReducer. El agregado sigue almacenándose en result_key, y las claves que devuelve el reductor se combinan con el estado del flujo de trabajo:
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,
];
}
}
La salida del reductor se combina con el estado del flujo de trabajo como la salida de cualquier otro paso. También puede devolver _next_step para crear ramas después de que termine el bucle para cada elemento.
Comportamiento de los reintentos
Las ramas del bucle para cada elemento se reintentan de forma independiente según config() del controlador de la rama. Si falla el paso del bucle y llamas a retry_workflow(), Queuety vuelve a expandir el paso y vuelve a poner en cola solo las ramas que aún faltan o han fallado. Los resultados de las ramas completadas se conservan en el estado del flujo de trabajo.
Cómo elegir entre parallel() y for_each()
Usa parallel() cuando la lista de ramas esté fijada en el código.
Usa for_each() cuando la lista de ramas se descubra durante la ejecución a partir del estado del flujo de trabajo y necesites modos de finalización como el primer éxito o el quórum.