Hello guys. I started implementing an interop modu...

# arrow-contributors

[Arkadii Ivanov]

Hello guys. I started implementing an interop module for Reaktive. Checked existing implementations for RxJava2 and Reactor. The CororutineContext->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]

Hi @Arkadii Ivanov,

That’s awesome. I’d love to pick your brain about MPP for Arrow Fx 😅.

That’s probably a bug in that case, the RxJava & Reactor implementations were done on a best-effort basis by me & @pakoito. If you see any other room for improvement that’d be awesome!

[simon.vergauwen]

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]

Looking very much forward to your contribution! Reaktive is awesome!

[Arkadii Ivanov]

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

[Arkadii Ivanov]

Will doo all my best

[simon.vergauwen]

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

[Arkadii Ivanov]

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

[simon.vergauwen]

Why would that be required?

[simon.vergauwen]

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

[Arkadii Ivanov]

From first point of view Channels of Flow will be required

[Arkadii Ivanov]

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]

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

[Arkadii Ivanov]

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

[Arkadii Ivanov]

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

[simon.vergauwen]

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]

Corotuine should suspend waiting for a task

[simon.vergauwen]

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

[Arkadii Ivanov]

If drop then how should I implement continueOn?

[simon.vergauwen]

continueOn is simply subscribeOn observeOn

[Arkadii Ivanov]

continueOn accepts CoroutineContext

[Arkadii Ivanov]

So we need interop

[simon.vergauwen]

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

[simon.vergauwen]

Right, we do need it

[Arkadii Ivanov]

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

[Arkadii Ivanov]

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]

Will take a while)

[simon.vergauwen]

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]

They also have the same priority and synchronisation problem.

[Arkadii Ivanov]

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