Как выполнить Zip (s1, o1), где s1 происходит только один раз, а o1 - несколько раз?

Question

Допустим, мне нужно сделать, когда s1 и o1 завершены, но s1 происходит только один раз, а o1 происходит несколько раз, и мне нужно что-то делать каждый раз, когда происходит o1. Как я могу это сделать?

Я попробовал следующий код,

val s1 = Single.fromCallable {
    Log.d("so","executing s1");
    "hello";
}.cache();

var emitter:ObservableEmitter<String>? = null;
val o1 = Observable.create<String> {
    emitter = it;
}

Observable.zip(s1.toObservable(),o1,
        BiFunction<String, String, Array<String>>() { t1, t2 -> arrayOf(t1,t2); })
        .subscribe { t1 ->
            Log.d("so", "Result = " + t1.joinToString(","));
        }

emitter?.onNext("homer");
emitter?.onNext("marge");
emitter?.onNext("bart");
Log.d("so", "Done");

Но результат был

executing s1,
Result = hello,homer
Done

Когда результат, который я ожидал, был

executing s1,
Result = hello,homer
Result = hello,marge
Result = hello,bart
Done

Answer 1

Использование flatMap:

s1.flatMapObservable({ t1 -> o2.map({ t2 -> arrayOf(t1, t2) }) })

Также используйте PublishSubject вместо того, чтобы красть излучатель:

val o2 = PublishSubject.create();

s1.flatMapObservable({ t1 -> o2.map({ t2 -> arrayOf(t1, t2) }) })
.subscribe { t1 ->
    Log.d("so", "Result = " + t1.joinToString(","));
}

o2.onNext("homer");
o2.onNext("marge");
o2.onNext("bart");

Log.d("so", "Done");