CombineLatest и холодный наблюдаемые

Question

Учитывая следующий код:

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

Я ожидал бы, чтобы напечатать что-то вроде следующего:

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

Но он печатает

3:a 
3:b 

Почему мой first Observable only, испускающий последний элемент? Активных подписок до combLatest нет, поэтому должно быть холодно. Как я могу убедиться, что все предметы из обоих наблюдаемых объединены вместе?

Answer 1

Это происходит потому, что у вас есть синхронные источники. Когда combineLatest подписывается на свои источники, он по умолчанию берет 128 элементов по умолчанию. Это означает, что ваш первый источник будет работать до завершения синхронно, а вторая подписка не произойдет до завершения. Поскольку первый источник один, все, кроме самого последнего элемента, удаляются. Как только второй источник подписывается, он найдет самое последнее из первого и объединится с ним.

Что вы пытались достичь? Если вы хотите попарно, используйте zip. Если вы хотите, чтобы все комбинации первого и второго, использовать flatMap:

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

Answer 2

Спасибо за ваш ответ, akarnokd. Предварительная выборка из 128 пунктов - очень полезная информация. Я решить мою проблему, переключая два наблюдаемые, так

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

теперь работает следующим образом:

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