Etapas paralelas

Às vezes, várias etapas podem ser executadas ao mesmo tempo. O método ->parallel() despacha um grupo de manipuladores de forma concorrente. O fluxo de trabalho só avança quando todos os manipuladores paralelos são concluídos.

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

Neste exemplo:

1. FetchUserHandler é executado primeiro (sequencialmente)
2. Três manipuladores são executados de forma concorrente (em paralelo)
3. MergeAndSaveHandler é executado depois que os três são concluídos (sequencialmente)

Como funciona

Quando uma etapa paralela é alcançada, Queuety despacha um trabalho para cada manipulador do grupo. Cada trabalho recebe o estado atual do fluxo de trabalho e é executado de forma independente. O fluxo de trabalho acompanha quantos trabalhos paralelos ainda estão em execução e só avança para a próxima etapa quando todos são concluídos.

O valor retornado por cada manipulador paralelo é mesclado ao estado do fluxo de trabalho. Se vários manipuladores retornarem a mesma chave, prevalecerá aquele que terminar por último.

Mesclagem do estado

Cada manipulador paralelo retorna dados que são mesclados ao estado compartilhado do fluxo de trabalho:

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

Após a conclusão da etapa paralela, o estado contém social_profile e payment_history, e a próxima etapa sequencial pode acessar tudo isso.

Combinação de etapas sequenciais e paralelas

Você pode combinar livremente chamadas ->then() e ->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 );

Tratamento de erros

Se algum manipulador de um grupo paralelo falhar, o fluxo de trabalho segue a lógica normal de novas tentativas para esse manipulador. Os outros manipuladores paralelos que já foram concluídos não são executados novamente. Quando o manipulador com falha é executado com sucesso, o fluxo de trabalho prossegue. Se ele esgotar as tentativas e for enterrado, o fluxo de trabalho será marcado como com falha e os trabalhos dos ramos restantes serão enterrados; use retry_workflow() para executar novamente a etapa com falha.