Rx - How to merge Single and Observables using Rx?

Question

I'm trying to create a Rx Chain so that I can have an Observable that can emit items, before and after executing a Single that will emit the same type of items.

Please check the example below:

    fun main() {
        val namesObservable = Observable.just(listOf("A", "B", "C"), listOf("D", "E", "F"))
        namesObservable
            .flatMapSingle { names -> addCharSingle(names) }
            .subscribe { item ->
                println(item.joinToString())
            }
    }

    fun addCharSingle(letters: List<String>): Single<List<String>> {
        return Single.fromCallable { letters.map { letter -> "$letter!" } }
    }

This implementation will print: "A!, B!, C!", "D!, E!, F!"

But I want it to print: "A, B, C", "A!, B!, C!", "D, E, F", "D!, E!, F!"

How can I do that?

Answer 1

Here is a way to do it:

1. Instead of flatMapSingle, use flatMap
2. Inside, use two Singles concatenated with each other.

• The first would have the original list, the second would do the conversion.

3. Lastly, convert it to Observable because Single.concatWith returns Flowable by design.

val namesObservable = Observable.just(
      listOf("A", "B", "C"), listOf("D", "E", "F"))

namesObservable
    .flatMap { names -> 
         Single.just(names)
            .concatWith(addCharSingle(names))
            .toObservable()
    }
    .subscribe { item ->
        println(item.joinToString())
    }