CombineLatest und kalt Observablen

Question

mit dem folgenden Code:

 Observable<String> obs1 = Observable.just("1", "2", "3"); 
     Observable<String> obs2 = Observable.just("a", "b"); 
     Observable.combineLatest(obs1, 
           obs2, 
           (s1, s2) -> s1 + ":" + s2 
     ).subscribe(System.out::println); 

ich erwarten würde es so etwas wie die folgenden drucken:

1:a 
2:a 
3:a 
3:b 

Aber es druckt

3:a 
3:b 

Warum ist erste Observable nur die letzte Position emittieren? Vor combinateLatest gibt es keine aktiven Abonnements, daher sollte es kalt sein. Wie kann ich sicherstellen, dass alle Objekte aus beiden Observablen miteinander kombiniert sind?

Answer 1

Dies geschieht, weil Sie synchrone Quellen haben. Wenn combineLatest seine Quellen abonniert, ruft es standardmäßig 128 Elemente von ihnen ab. Dies bedeutet, dass Ihre erste Quelle synchron zur Fertigstellung ausgeführt wird und das zweite Abonnement nicht einmal bis zu diesem Abschluss stattfindet. Da die erste Quelle alleine ist, werden alle außer dem allerletzten Element gelöscht. Sobald die zweite Quelle abonniert ist, findet sie die allerletzte von der ersten und kombiniert sich damit.

Was haben Sie versucht zu erreichen? Wenn Sie paarweise kombinieren möchten, verwenden Sie zip. Wenn Sie alle Kombinationen von ersten und zweiten möchten, verwenden Sie flatMap:

obs1.flatMap(v -> obs2.map(w -> v + ":" + w)).subscribe(...) 

Answer 2

Vielen Dank für Ihre Antwort, akarnokd. Der Prefetch von 128 Elementen ist ein sehr nützlicher Einblick. Ich habe mein Problem durch Umschalten der beiden Observablen gelöst, so

Observable<String> obs1 = Observable.just("1", "2", "3", "4"); 
Observable<String> obs2 = Observable.just("4", "b", "1"); 
Observable.combineLatest(obs1, 
         obs2.toList(), 
         (s1, list) -> list.stream().anyMatch((listItem) -> s1.contains(listItem)) ? null : s1) 
      .filter(StringUtils::isNotEmpty) 
      .subscribe(System.out::println); 

arbeitet nun wie folgt:

Observable<String> obs1 = Observable.just("1", "2", "3", "4"); 
Observable<String> obs2 = Observable.just("4", "b", "1"); 
Observable.combineLatest(obs2.toList(), 
         obs1, 
         (list, s1) -> list.stream().anyMatch((listItem) -> s1.contains(listItem)) ? null : s1) 
      .filter(StringUtils::isNotEmpty) 
      .subscribe(System.out::println);