using Rxjava with flatmap not change the downstream data

Question

RxJava3
Kotlin

I have the following method and I am trying to control the downstream to return the Pair<Customer, Profile> after the map operator

However, after the first flatMap the data is changes to customer. Basically, I want to return the Pair<Customer, Profile> and not the customer dowwstream.

override fun execute(): Single<Pair<Customer, Profile>> {
        return Single.fromObservable(
            graphQLProfileService.getCustomerProfile()
        ).map {
            Pair(
                it.mdc.toCustomer(),
                it.profile
            )
        } /* Single<Pair<Customer, Profile>> */
            .flatMap { /* it: Pair<Customer, Profile> */
                updateProfileUseCase.execute(topsCustomer = it.first)
            }
            .flatMap { /* it: Customer */
                updateUserUseCase.execute().toSingleDefault(it) /* it.first, as the first item will be customer */
            }
            .flatMap { /* it: Customer */
                updateAppUseCase.execute().toSingleDefault(it) /* it.first */
            }
}

I think the solution should be something like this but I haven't managed to get it to work: https://medium.com/@douglas.iacovelli/rxjava-kotlin-keep-the-original-value-in-a-flatmap-bbd6a6974a99

Answer 1

You have to put those flatMaps inside a flatMap and once they succeed, you have to replace their results back to the value by ignoring it and concatenating with a just of the pair:

Single.fromObservable(
    graphQLProfileService.getCustomerProfile()
).map {
    Pair(
        it.mdc.toCustomer(),
        it.profile
    )
}
.flatMap { original ->
    Single.just(original) // <--------------------------------------
    .flatMap { /* it: Pair<Customer, Profile> */
        updateProfileUseCase.execute(topsCustomer = it.first)
    }
    .flatMap { /* it: Customer */
        updateUserUseCase.execute().toSingleDefault(it) /* it.first, as the first item will be customer */
    }
    .flatMap { /* it: Customer */
        updateAppUseCase.execute().toSingleDefault(it) /* it.first */
    }
    .ignoreElement() // <--------------------------------------
    .andThen(Single.just(original))
}