CombineLatest nie subskrybuje parametru Observables, nie dostarcza danych - kątowe, rxjs, obserwowalne, combinelatest

Mam dość powszechny przypadek użycia. Ładuję tablicę obiektów z FirebaseDatabase i dołączam obrazy z Firebase Storage.

Baza danych FirebaseDatabase zwraca obserwowalne na podstawiepytanie. Za każdym razem dane w bazie danych ulegną zmianie, obserwator wygeneruje zaktualizowany wynik zapytania. Dołączam również Promise z api Storage do każdego wpisu w wyniku poprzez mapowanie do krotki [Domain, ImagePromise] dla każdego wpisu

getDomainObj(){
return this.af.database.list(this.listRef)
.map((mps: Domain[]) =>
mps.map((mp: Domain) =>
[mp, this.getImage(mp["$key"])]));
}

Dwie kwestie:

1. niektórzy mps nie mają obrazu.
2. Ponieważ obrazy nie są w firedb, toobserwowalny nie będzie emitował przy zmianie obrazu. Aby to naprawić, dodałem funkcję chmury, aby wysłać powiadomienie push przy każdej aktualizacji obrazu (poza zakresem tego pytania działa, więc nie trzeba w to wchodzić). Każdy MP to osobny temat PubSub, a my, klient, subskrybujemy ten temat i otrzymujemy obserwowalne aktualizacje, które będą emitować za każdym razem, gdy zostanie przesłany obraz dla określonego mp.

Rozwiązuję je poprzez mapowanie

this.mpSvc.getDomainObj()
.map((tupleArray: [Domain, Promise<string>][]) =>
tupleArray.map(([mp, imageSrcPromise]: [Domain, Promise<string>]) => {
return [mp,
Observable.fromPromise(imageSrcPromise) // get initial image
.startWith(null) // make sure that it emits at least once
.concat( // updates
this.mpSvc.signUpForPictureUpdates(mp["$key"])
.map(([id, imageSrc]: [string, string]) => imageSrc))

];
})
)

mpSvc.signUpForPictureUpdates zwraca krotki Observable of domainId-imageSrc, które będą emitować za każdym razem, gdy obraz domeny zostanie ponownie załadowany

Teraz ostatecznym celem jest posiadanie obserwowalnegotablica [Domain, imageSrc], która będzie emitować za każdym razem, gdy zmieni się obraz dowolnego obiektu domeny lub dowolna zmiana obiektu domeny lub wynik zmiany zapytania.

Najpierw ponownie remapuję, zamiast emitować tablicę [Domain, Observable] par, emitować tablicę Observable <[Domain, imageSrc]>

.map((tupleArray: [Domain, Observable<string>][]) =>
tupleArray.map(([mp, imageSrcObservable]: [Domain, Observable<string>]) =>
imageSrcObservable.map((imageSrc: string) => [mp, imageSrc])))

teraz mam Obserwowalny szeregObserwowalne par. Parent Observable będzie emitował za każdym razem, gdy zmieni się Domain lub wynik zapytania, a dzieci będą emitować za każdym razem, gdy zmieni się obraz dla określonej domeny.

Ostatnim krokiem jest połączenie obserwowalnych dzieci w jedno obserwowalne i przełączanie do niego.

.switchMap((imageObservables: Observable<[Domain, string]>[]) =>
Observable.combineLatest(imageObservables)
)

wynik jest poprawnie zapisany (w rzeczywistości jest wyświetlany w szablonie kątowym przez ngFor | async, ale to również jest poza zakresem).

Niestety, wyniki nie są zgodne z oczekiwaniami. W rzeczywistości mam cztery obiekty wynikowe z zapytania, dwa z obrazami i dwa bez. To, co widzę, nie jest spójne - czasami ładują się oba obrazy, czasem jeden, a najczęściej nie.

Konkretnie, gdybym miał dodać wiersz dziennika na końcu, tak:

.do(x => console.log("final", x), x => console.error("finalerror", x), () => console.log("finalcomplete"));

Zawsze otrzymuję co najmniej jedną linię dziennika z wartościami null w imageSrc, ale prawie nigdy nie pojawiają się dodatkowe linie z rozwiązanymi prawdziwymi imageSrcs.

subskrybowanie przed ostatnim krokiem (CombineLatest) pobiera wszystkie dane poprawnie

Co ja robię źle?

Edycja: po rozpatrzeniu tego więcej, problem jest z pewnością z combineLatest. Próbowałem wymienić combineLatest z imageObservables[1].map(x=>[x]) i działa idealnie (choć oczywiście tylkozwraca jedną wartość zamiast tablicy. Również przejście przez CombineLatest doprowadziło mnie do jakiegoś zamknięcia, które jest dziwne, ponieważ to, co musiałoby być zamknięte w dowolnym momencie?

Odpowiedzi:

2 dla odpowiedzi № 1

Ok, znalazłem problem. fromPromise rzuci błąd, jeśli Promise odrzuci. Tak więc jeden z błędów dzieci "wyskoczył, kiedy to getImage nie dostał obrazu, który powoduje combineLatest ukończyć.

Naprawiłem to w ten sposób

Observable.fromPromise(imageSrcPromise) // get initial image
.catch(() => Observable.empty()) // <-- fix
.startWith(null) // and so forth...

Dlatego czasami emitował poprawnie, iczasem nie było. Jeśli żądania obiektów Domain z obrazami załadowanymi jako pierwsze, byłyby wyświetlane, a jeśli obiekty bez obrazu zostały zwrócone jako pierwsze, strumień by się skończył.

---