Rxjava with kotlin coroutines

Question

I am in need that i need to start the coroutines from rxkotlin chain, But am i not sure whether this is right or wrong, to start a coroutines from the rx chain i use runblocking to start the suspend methods
Example

Single.just(someOperation())
   .map{
     someMethod(it)
   }
  .flatMap{
    startCoroutines(suspend { someOpeartions() } ) // i will be starting the coroutines here
  }

Coroutines

fun startCoroutines(suspendingObj : suspend () -> Any){
  runBlocking(newFixedThreadPoolContext(1,"Thread")){
       suspendingObj.invoke()
  }
}

Is this above code is correct way of doing it or is there is any other way to achieve this ? Can anyone help me out with this

Why do you need coroutines if it's already inside the chain? I guess it should either rx or coroutines. — Tenten Ponce, Jul 19 '19 at 05:06
rx is already asynchronous, why do you want to use coroutines inside it? if your goal is to change thread, it is already supported on rx `.observeOn()`. You can just instead of suspend, just make it rx and add it on your current chain. — Tenten Ponce, Jul 19 '19 at 05:09
Blocking is almost never a good choice, Rx or no Rx. [There exist](https://github.com/Kotlin/kotlinx.coroutines/tree/master/reactive/kotlinx-coroutines-rx2) converters and builders that can bridge the two approaches. For example [rxSingle](https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-rx2/kotlinx.coroutines.rx2/kotlinx.coroutines.-coroutine-scope/rx-single.html). — akarnokd, Jul 19 '19 at 08:13

score 1 · Accepted Answer · answered Jul 27 '19 at 11:51

This code chunk is fundamentally wrong.

In your case it is really not necessary to use Coroutines, because you can easily change Rx thread before flatMap with observeOn and pass any Scheduler you want (like IO).
Kotlin Coroutines were designed to avoid Threads, because creating Threads is a very expensive operation. And your function startCoroutines creates a new thread for every operation which makes no sense and potentially will cause overflow. You can read more about it here: Difference between a "coroutine" and a "thread"?
You should always try to find a better system design before deciding using runBlocking. Blocking threads is never a good idea.

Rxjava with kotlin coroutines

1 Answers1