Към съдържанието

Изпълнение на работни потоци

Преглед като Markdown Редактиране в GitHub Документация за 2cc7b2b

Работният поток може да стартира дъщерен работен поток като една от своите стъпки. Родителският поток се поставя на пауза, докато дъщерният поток не завърши, след което крайното му състояние се обединява със състоянието на родителския поток.

Методът ->run_workflow()

PHP
use Queuety\Queuety;

$enrichment = Queuety::workflow( 'enrich_user' )
    ->then( FetchSocialHandler::class )
    ->then( FetchCreditScoreHandler::class );

Queuety::workflow( 'onboarding' )
    ->then( CreateAccountHandler::class )
    ->run_workflow( 'enrich_user', $enrichment )
    ->then( SendWelcomeEmailHandler::class )
    ->dispatch( [ 'email' => 'user@example.com' ] );

Методът run_workflow() приема два аргумента:

  1. Име на работния поток за изпълнение
  2. Екземпляр WorkflowBuilder, който задава стъпките му

Как работи

Когато родителският работен поток достигне стъпка run_workflow:

  1. Работният поток за изпълнение се изпраща като независим работен поток със собствен идентификатор
  2. Родителският работен поток се поставя на пауза
  3. Работният поток за изпълнение преминава обичайно през стъпките си
  4. След завършването му крайното му състояние се обединява със състоянието на родителския поток
  5. Родителският работен поток продължава със следващата стъпка

Състояние на родителския и дъщерния поток

Работният поток за изпълнение получава текущото натрупано състояние на родителския поток като начален полезен товар. Така стъпките в дъщерния поток имат достъп до данните, създадени от предходните стъпки на родителския поток:

PHP
class CreateAccountHandler implements Step {
    public function handle( array $state ): array {
        $user_id = wp_create_user( $state['email'], wp_generate_password() );
        return [ 'user_id' => $user_id ];
    }

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

// This run-workflow step can access 'user_id' from the parent
class FetchSocialHandler implements Step {
    public function handle( array $state ): array {
        $profile = fetch_social_profile( $state['user_id'] );
        return [ 'social_data' => $profile ];
    }

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

Обединяване на състоянието

Когато работният поток за изпълнение завърши, крайното му състояние (без вътрешните ключове като _steps, _queue, _priority, _max_attempts) се обединява със състоянието на родителския поток. Следващата стъпка в родителския поток вижда всички данни от предходните му стъпки и от работния поток за изпълнение:

PHP
class SendWelcomeEmailHandler implements Step {
    public function handle( array $state ): array {
        // Has access to:
        // - $state['email'] (from initial dispatch)
        // - $state['user_id'] (from CreateAccountHandler)
        // - $state['social_data'] (from run-workflow's FetchSocialHandler)
        // - $state['credit_score'] (from run-workflow's FetchCreditScoreHandler)

        send_welcome_email( $state['email'], $state['social_data'] );
        return [ 'welcome_sent' => true ];
    }

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

Опции за работния поток за изпълнение

on_queue(), with_priority() и max_attempts() на конструкторa за работния поток за изпълнение се прилагат към дъщерния работен поток. Другите опции на ниво работен поток в дъщерния конструктор, като on_cancel(), prune_state_after(), must_complete_within(), version(), idempotency_key(), бюджетите max_*() и compensate_on_failure(), не се пренасят:

PHP
$sub = Queuety::workflow( 'heavy_processing' )
    ->on_queue( 'background' )
    ->with_priority( Priority::Normal )
    ->max_attempts( 5 )
    ->then( StepA::class )
    ->then( StepB::class );

Queuety::workflow( 'parent' )
    ->run_workflow( 'heavy', $sub )
    ->then( FinalStep::class )
    ->dispatch( $payload );

Вложени работни потоци за изпълнение

Работните потоци за изпълнение могат сами да съдържат такива потоци, което позволява произволно влагане:

PHP
$inner = Queuety::workflow( 'inner' )
    ->then( InnerStepA::class )
    ->then( InnerStepB::class );

$middle = Queuety::workflow( 'middle' )
    ->then( MiddleStepA::class )
    ->run_workflow( 'inner', $inner )
    ->then( MiddleStepB::class );

Queuety::workflow( 'outer' )
    ->then( OuterStepA::class )
    ->run_workflow( 'middle', $middle )
    ->then( OuterStepB::class )
    ->dispatch( $payload );