angular - Bir gözlemlenebilirler listesini veya bunların aboneliklerini doğru şekilde forkJoin nasıl yapılır?

Paralel olarak çalışması gereken çok büyük bir sıralı gözlemler listem var. Her bir gözlem döndüğünde, sonuçlarını bir davranış konusuna ekliyor, böylece sonuçlar iletiliyor. Ancak, hepsi tamamlandığında çağrılması gereken özel bir fonksiyona ihtiyacım var.

Her bir gözlemlenebilir, bir API'den bir resim (ve ilgili metadata) indirir. İsteklerin mümkün olan en hızlı şekilde gerçekleştirilmesi gerekiyor ve her sonuç yayımlandığında onunla çalışmam ve tüm gözlemlenebilirler tamamlandığında boş bir değer yaymam gerekiyor. Bu, gözlemlenebilirlerin paralel olarak yürütülmesi gerektiği anlamına gelir.

Orijinal uygulama, tamamlandıklarında bir geri çağırma olmadan.

const requests: Observable[] = getRequests();

requests.forEach(obs => obs.subscribe(res => {
    const currentImages = this.$images.value;
    currentImages.push(res);
    this.$images.next(currentImages);
}));

Tüm istekler tamamlandığında geri aramayı uygulamak için aşağıdakileri denedim.

const requests: Observable[] = getRequests();
const finishedTracker = new Subject<void>();

requests.forEach(obs => obs.subscribe(res => {
    const currentImages = this.$images.value;
    currentImages.push(res);
    this.$images.next(currentImages);
}));

forkJoin(requests).subscribe(() => {
    finishedTracker.next();
    finishedTracker.complete();
    console.log('requests done');
});

Bu işe yarıyor, ama bana garip geliyor ki forkJoin ve isteklere yapılan abonelikleri ayırmam gerekiyor. Bu işlevselliği uygulamanın daha iyi bir yolu var mı? mergeMap'e de baktım ama çalıştırmayı başaramadım.

Düzenle Yorumlara dayanarak, iki kez abone olmanın, talepleri iki kez yapmak anlamına geldiğini fark ettim. Bu yüzden başka bir uygulama denedim.

from(requests).pipe(
    mergeMap(o => {
        o.subscribe(res => {
            const currentImages = this.$images.value;
            currentImages.push(res);
            this.$images.next(currentImages);
        }
        return o;
    }, 10)
).subscribe(() => {
    finishedTracker.next();
    console.log('requests done');
})

forkJoin'den elde edilen sonucu kullanmadım çünkü anladığım kadarıyla bana tüm isteklerin sonucunu veriyor. Bu nedenle bunların tamamlanmasını beklemesi gerekiyor. Her bir istek nispeten hızlı olmasına rağmen, çoğu zaman yüzlercesi oluyor ve her isteğin bitmesiyle birlikte bireysel sonuçlarının davranış konusuna iletilmesi gerekiyor.

Düzenleme 2 Benim tercih ettiğim çözüm.

from(requests).pipe(
    mergeMap(request => request, 10),
    scan<ImageResponse, ImageResponse[]>((all, current, index) => {
        all = all.concat(current);
        this.$images.next(all);
        return all;
    }, [])
).subscribe({
    complete: () => {
    finishedTracker.next();
    console.log('requests done');
}});

Cevap #1

mergeMap içindeki abone olmanıza gerek yok. Aslında, başkalarının da belirttiği gibi, bu çift aboneliğe neden oluyor çünkü mergeMap, ona geçirdiğiniz fonksiyonun döndürdüğü gözlemlenebilir objeye dahili olarak abone oluyor.

Yanıtları gerçekleştiği gibi ele almak için basitçe bir boru kullanabilir ve işleme mantığınızı içine ekleyebilirsiniz. Temelde bir yan etki (mevcut akışın çıktısını değiştirmeyen bir şey) yaptığınız için, tap operatörünü kullanmak uygundur:

from(requests).pipe(
    mergeMap(o => o.pipe(
        tap(res => {
            const currentImages = this.$images.value;
            currentImages.push(res);
            this.$images.next(currentImages);
        }),
    }, 10)
).subscribe(() => {
    finishedTracker.next();
    console.log('requests done');
})

Bu işe yarayacak olsa da, gözlemlenebilir akışı fazla karmaşık hâle getirdiğiniz görünüyor. Kullanım durumunuzdan tam olarak emin değilim, ancak tahminimce Subjects aslında hiç gerekli değil. Amacınız, işlenen sonuçların kümülatif bir dizisini yayınlamaksa, bunu herhangi bir Subject veya BehaviorSubject kullanmadan scan ile yapabilirsiniz. Tüm istekler tamamlandığında bazı mantıklar çalıştırmak için, yalnızca complete geri çağrısını belirten kısmi bir Observer iletilebilir (bu, subscribe()'e bir fonksiyon ilettiğinizde varsayılan olarak kullanılan next geri çağrısı yerine):

from(requests).pipe(
    mergeMap(request => request, 10),
    scan((all, current) => all.concat(current), [])
).subscribe({
    complete: () => console.log('requests done')
});

DÜZENLEME: @AdrianBrand tarafından belirtildiği gibi, from/mergeMap yerine merge kullanmak daha özlüdür:

merge(...requests, 10).pipe(
    scan((all, current) => all.concat(current), [])
).subscribe({
    complete: () => console.log('requests done')
})

Cevap #2

Basit bir combine latest hakkında ne düşünüyorsun?

import { of, Observable, combineLatest, startWith } from 'rxjs';

declare const getRequests: () => Observable<any>[];

const requests: Observable<any>[] = getRequests();


combineLatest(requests.map(r => r.pipe(startWith(null)))).subscribe(result => {

  if (!result.some(r => r === null)) {
    // everything loaded
  } else {
    // not finished
  }
})

Oyun alanı

İlgili makaleler