For-Each Dinamis

Lihat sebagai Markdown Edit di GitHub Dokumentasi di 2cc7b2b

Gunakan ->for_each() saat alur kerja belum mengetahui jumlah cabang pada waktu build. Queuety membaca array dari status alur kerja, mengirim satu tugas cabang untuk setiap item, dan melanjutkan hanya setelah kondisi penyelesaian yang dikonfigurasi terpenuhi.

Metode `->for_each()`

PHP

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

Langkah perencana menulis pages ke dalam status alur kerja. Langkah for-each membaca array tersebut dan mengirim satu tugas cabang untuk setiap item.

Handler cabang

Handler cabang for-each mengimplementasikan Contracts\ForEachHandler:

PHP

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',
        ];
    }
}

Di config(), backoff menerima nama strategi: exponential, linear, atau fixed. Daftar penundaan per percobaan di sana saat ini diabaikan dan menjadi percobaan ulang langsung (#16); letakkan penundaan per percobaan di backoff milik definisi langkah.

$state berisi seluruh status alur kerja, $item adalah item for-each, dan $index adalah posisinya yang dimulai dari nol.

Mode penyelesaian

ForEachMode::All

Langkah hanya selesai setelah setiap cabang berhasil. Kegagalan permanen pada cabang mana pun membuat langkah for-each gagal.

ForEachMode::FirstSuccess

Langkah selesai segera setelah cabang pertama berhasil. Cabang aktif yang tersisa ditandai sebagai digantikan.

ForEachMode::Quorum

Langkah selesai saat setidaknya quorum cabang berhasil. Jika cabang yang tersisa tidak lagi dapat memenuhi kuorum, alur kerja gagal.

Reducer

Secara default, Queuety menyimpan agregat terstruktur di result_key:

mode
quorum
total
succeeded
failed
settled
winner
results
failures

Jika Anda menginginkan ringkasan khusus, berikan kelas reducer yang mengimplementasikan Contracts\ForEachReducer. Agregat tetap disimpan di result_key, dan kunci yang dikembalikan reducer digabungkan ke status alur kerja di sebelahnya:

PHP

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

Output reducer digabungkan ke status alur kerja seperti output langkah lainnya. Reducer juga dapat mengembalikan _next_step untuk melakukan percabangan setelah for-each selesai.

Perilaku percobaan ulang

Cabang for-each dicoba ulang secara independen sesuai dengan config() handler cabang. Jika langkah for-each gagal dan Anda memanggil retry_workflow(), Queuety akan memperluas kembali langkah for-each dan hanya memasukkan kembali cabang yang masih belum ada atau gagal ke antrean. Hasil cabang yang berhasil tetap berada dalam status alur kerja.

Memilih antara `parallel()` dan `for_each()`

Gunakan parallel() saat daftar cabang ditetapkan dalam kode.

Gunakan for_each() saat daftar cabang ditemukan saat runtime dari status alur kerja dan Anda memerlukan mode penyelesaian seperti keberhasilan pertama atau kuorum.