Есть ли в RxJava2 какая-либо функция, например CombineLatest, которая не ожидает событий от обоих наблюдателей?

Question

Ближе всего к тому, что я хочу, функция combineLatest выполняет следующее:

Но мне нужно это:

Я посмотрел почти все функции RxJava, но не увидел ничего подходящего.

Answer 1

Вы можете добавить «пустой» элемент в начале, поэтому "1" + "" = "1":

    Subject<String> two = PublishSubject.create();
    Subject<Integer> one = PublishSubject.create();

    Observable
            .combineLatest(one, Observable.just("").concatWith(two), (i, s) -> "" + i + s)
            .subscribe(s -> System.out.print(s + " "));

    one.onNext(1);
    one.onNext(2);
    two.onNext("A");
    two.onNext("B");
    two.onNext("C");
    two.onNext("D");
    one.onNext(3);
    one.onNext(4);
    one.onNext(5);

печатает:

1 2 2A 2B 2C 2D 3D 4D 5D 

Answer 2

Спасибо @zella и @Gustavo за их решения, это почти то, что мне нужно, но в их случаях объединенный источник генерирует только событие одного из источников. Так что фрагмент ниже делает именно то, что я хочу:

Observable.combineLatest(
    source1.startWith(""),
    source2.startWith(""),
    BiFunction<String, String, String> { s1, s2->
        s1 + s2
    })