Hello guys I started implementing an interop module for Reak

Question

Hello guys I started implementing an interop module for Reaktive Checked existing implementations for RxJava2 and Reactor The CororutineContext gt Scheduler converter appears to be implemented incorrectly E g tasks scheduled on a Worker are guaranteed to be executed sequentially and in a non overlapping fashion but here they are being executed in parallel <https github com arrow kt arrow blob 397b84f1366662c8a0f4c3ea02717e1c74ccc990 modules fx arrow fx rx2 src main kotlin arrow fx rx2 CoroutineContextRx2Scheduler kt L29> And I believe same issue in Reactor module So I m wondering is there any reason of such an implementation or it s just a bug

simon.vergauwen · Answer

Hi < Arkadii Ivanov> That s awesome I d love to pick your brain about MPP for Arrow Fx sweat smile That s probably a bug in that case the RxJava amp Reactor implementations were done on a best effort basis by me amp < pakoito> If you see any other room for improvement that d be awesome

simon.vergauwen · Answer

Reactor is also still missing `Concurrent` implementation for all its types `Bracket` which composes 3 of the same Rx types together while guaranteeing resource safety was especially hard I was surprised neither frameworks exposes such an operation they only support side effecting onces

simon.vergauwen · Answer

Looking very much forward to your contribution Reaktive is awesome

Arkadii Ivanov · Answer

Cool On the other hand I m completely new to Arrow lib

Arkadii Ivanov · Answer

Will doo all my best

simon.vergauwen · Answer

If you have any questions I d be happy to answer all of them or jump on a call if needed +1

Arkadii Ivanov · Answer

I think for correct implementation a dependency to kotlinx coroutines would be required

simon.vergauwen · Answer

Why would that be required

simon.vergauwen · Answer

In that case I personally would drop it from the module instead of introducing a dependency for a single combinator

Arkadii Ivanov · Answer

From first point of view Channels of Flow will be required

Arkadii Ivanov · Answer

So you need to start a coroutine and receive tasks one by one for execution I will think more later maybe there is a better solution

simon.vergauwen · Answer

We have `Queue` etc in Arrow Fx as well which is the FP equivalent of `Channel`

Arkadii Ivanov · Answer

Also delayed tasks should be prioritized so there should be kinda PriorityQueue

Arkadii Ivanov · Answer

```We have Queue etc in Arrow Fx as well which is the FP equivalent of Channel```

simon.vergauwen · Answer

Yes sounds we should drop it for now if it involves that Explains why `Schedulers computation ` was acting weird but `<http Schedulers io|Schedulers io> ` wasn t

Arkadii Ivanov · Answer

Corotuine should suspend waiting for a task

simon.vergauwen · Answer

Which is true for `Queue` in Arrow Fx It semantically blocks or suspends

Arkadii Ivanov · Answer

If drop then how should I implement `continueOn`

simon.vergauwen · Answer

`continueOn` is simply ~`subscribeOn`~ `observeOn`

Arkadii Ivanov · Answer

`continueOn` accepts `CoroutineContext`

Arkadii Ivanov · Answer

So we need interop

simon.vergauwen · Answer

It changes the threading for downstream which is `observeOn` and you can install multiple times

simon.vergauwen · Answer

Right we do need it

Arkadii Ivanov · Answer

E g we have a `Single` wrapped into `SingleK` The last one has `continueOn` operator I need to implement this operator

Arkadii Ivanov · Answer

Cool thanks guys for your support I m offered 20 of my work hours to spend on open source Plus a bit of spare time

Arkadii Ivanov · Answer

Will take a while

simon.vergauwen · Answer

Cool will help wherever I can I was just thinking creating a `Scheduler` from a `CoroutineContext` shouldn t be any different from creating a scheduler from an `ExecutorService` isn t it

simon.vergauwen · Answer

They also have the same priority and synchronisation problem

Arkadii Ivanov · Answer

Yup same problems in general That s why RxJava2 s `ExecutorScheduler` implementation is ~450 lines long And it uses tricky queue under the hood