Pasos paralelos

A veces, varios pasos pueden ejecutarse al mismo tiempo. El método ->parallel() envía un grupo de controladores de forma concurrente. El flujo de trabajo avanza solo cuando todos los controladores paralelos han terminado.

El método ->parallel()

use Queuety\Queuety;

Queuety::workflow( 'enrich_profile' )
    ->then( FetchUserHandler::class )
    ->parallel( [
        FetchSocialProfileHandler::class,
        FetchPaymentHistoryHandler::class,
        FetchActivityLogHandler::class,
    ] )
    ->then( MergeAndSaveHandler::class )
    ->dispatch( [ 'user_id' => 42 ] );

En este ejemplo:

1. FetchUserHandler se ejecuta primero, de forma secuencial
2. Tres controladores se ejecutan de forma concurrente, en paralelo
3. MergeAndSaveHandler se ejecuta cuando terminan los tres, de forma secuencial

Cómo funciona

Al llegar a un paso paralelo, Queuety envía un trabajo por cada controlador del grupo. Cada trabajo recibe el estado actual del flujo de trabajo y se ejecuta de forma independiente. El flujo lleva la cuenta de los trabajos paralelos pendientes y solo avanza al siguiente paso cuando todos han terminado.

El valor devuelto por cada controlador paralelo se combina con el estado del flujo de trabajo. Si varios controladores devuelven la misma clave, prevalece el último en terminar.

Combinación del estado

Cada controlador paralelo devuelve datos que se combinan con el estado compartido del flujo de trabajo:

class FetchSocialProfileHandler implements Step {
    public function handle( array $state ): array {
        $profile = fetch_social_api( $state['user_id'] );
        return [ 'social_profile' => $profile ];
    }

    public function config(): array {
        return [];
    }
}

class FetchPaymentHistoryHandler implements Step {
    public function handle( array $state ): array {
        $payments = fetch_payment_api( $state['user_id'] );
        return [ 'payment_history' => $payments ];
    }

    public function config(): array {
        return [];
    }
}

Cuando termina el paso paralelo, el estado contiene tanto social_profile como payment_history, y el siguiente paso secuencial puede acceder a todo.

Combinar pasos secuenciales y paralelos

Puedes combinar libremente las llamadas a ->then() y ->parallel():

Queuety::workflow( 'complex_pipeline' )
    ->then( SetupHandler::class )               // sequential
    ->parallel( [ TaskA::class, TaskB::class ] ) // parallel
    ->then( IntermediateHandler::class )         // sequential
    ->parallel( [ TaskC::class, TaskD::class ] ) // parallel
    ->then( FinalizeHandler::class )             // sequential
    ->dispatch( $payload );

Gestión de errores

Si falla algún controlador de un grupo paralelo, el flujo de trabajo sigue la lógica de reintento habitual para ese controlador. Los demás controladores paralelos que ya se hayan completado no se vuelven a ejecutar. Cuando el controlador fallido se ejecuta correctamente, el flujo de trabajo continúa. Si agota sus reintentos y queda en la cola de fallidos, el flujo de trabajo se marca como fallido y los trabajos restantes de la rama también pasan a la cola de fallidos; usa retry_workflow() para volver a ejecutar el paso fallido.