Coroutine based state store

[ATizik]

Left most of queues work intact, aggregated it inside an actor. Switched state subject to channel. Removed flush subject, now it's a responsibility of the actor. Added a way to synchronously get a value from actor(meaning that it's passed as message to and actor just like set and get messages, and is processed in same iterator)

Tests pass, but with artificial delays because I couldn't at this devise a method to make actor work synchronously for tests, I'll fix it later.

Extracted stateStore interface for tests.
Added concurrency test. Be advised that original MvRxStateStore rarely fails new concurrency test, that's why I left 1000 identical tests in a row to demonstrate.

Didn't find contributing guide, so I'll await feedback.
TODO for me: fix tests and document new code

UPDATE:
Fixed tests to run synchronously like before

UPDATE_2:
Simplified flushing logic. Now it iterates over every element of an actor's channel, placing every GetQueueElement to it's own queue and processing every SetQueueElement right away. Once channel is empty getQueue is processed

UPDATE_3:
Transition to coroutines 0.26.0 with structured concurrency changes

UPDATE_4:
Updated to Kotlin 1.3.0 and Coroutines 1.0.0

[gpeal]

Thank you @ATizik we're going to dig in and take a close look at this ❤️

[Dwite]

What do you think if we change DefaultDispatcher to newSingleThreadContext ? I'm not sure how the actor will work on Dispatcher with multiple threads

[ATizik]

If my understanding is correct, it doesn't matter for actors, since actor is a coroutine and is executed sequentially. Here is relevant documentation. It might make sense to create a separate cached pool though, because Default can be starved by other tasks

[gpeal]

@ATizik Do you want to update this to stable coroutines? Now that they are stable, it's time to take a serious look at this.

@BenSchwab @elihart @hellohuanlin

[ATizik]

@gpeal
Yep, I'll update MR in a few days

[ATizik]

Updated to Kotlin 1.3.0 and Coroutines 1.0.0

You might want to wait a bit more if you want to rely on stable api, because ConflatedBroadcastChannel is marked as @ExperimentalCoroutinesApi,
actor builder is marked as @ObsoleteCoroutinesApi, due to be replaced by complex actors:
Kotlin/kotlinx.coroutines#87

[Hesamedin]

Is there any update on this PR?
I am using Coroutines in my project and having this PR is great.
Thanks to all of you who made this possible 👍

[gpeal]

@Hesamedin this PR doesn't affect the external API of MvRx, just the internal implementation.

[Hesamedin]

Oh, I thought this is the fix for #130 :(
I am using Coroutines in my project for handling Network calls and I am wondering am I able to use Async the way you are using for Rx.execute()?

Anyways, thanks guys for building such an awesome library. I am using it and it and I love it 🥰 👍

[Dwite]

@telus you can use wrapper around Deferred or direct results from suspend function that will map your network calls to Async

[gpeal]

@Dwite @Hesamedin execute is just a convenience around RxJava and you could create your own extension function for coroutines. However, we may ship a separate mvrx-coroutines artifact in the future to make it work out of the box (tbd)
Something like this (untested code) should work:

fun <T : Any> CoroutineScope.execute(fn: suspend () -> T, dispatcher: Dispatcher = Dispatchers.Default, reducer: S.(Async<T>) -> S) {
    launch {
        setState { reducer(Loading()) }
       try {
           setState { reducer(Success(fn())) }
        } catch (e: Exception) {
            setState { reducer(Fail(e)) }
    }
}

[digitalbuddha]

my understanding is that jcenter is a supercet of mavenCentral do you need both?

[digitalbuddha]

I don't think this is needed, allprojects repos are being added in the root build.gradle

[digitalbuddha]

why add build script repositories at the module level?

[digitalbuddha]

please upgrade to latest 1.3.40

[digitalbuddha]

rather than having the class implement CoroutineScope could you have it own a coroutine scope as a private property. Android did a similar thing with fragment.lifecycleScope. Think of it as not repeating the problem of an activity being a context but rather an activity owning a context.

[ATizik]

It's just for convenince since every coroutine expects to be started in CoroutineScope, and this class should be passed around by it's interface so it doesn't expose it's scope. Also in kotlin documentation activity implements coroutine scope through delegate. If you want I'll refactor to private property, I don't have a strong opinion on which way is the right one

[digitalbuddha]

if flipping to a coroutine store, would it make sense to go all the way and change observable to flow? cc @gpeal & @BenSchwab

[digitalbuddha]

I'm not perfectly primed on actors yet, what is the advantage of using an actor rather than a straight channel or flow here?

[ATizik]

It's basically a shorhand for this:

private val queueChannel = Channel<Job<S>>(capacity = Channel.UNLIMITED)
private val actor = launch {
   queueChannel.consumeEach {

From documentation:
An actor is an entity made up of a combination of a coroutine, the state that is confined and encapsulated into this coroutine, and a channel to communicate with other coroutines. A simple actor can be written as a function, but an actor with a complex state is better suited for a class.

There is an actor coroutine builder that conveniently combines actor's mailbox channel into its scope to receive messages from and combines the send channel into the resulting job object, so that a single reference to the actor can be carried around as its handle.

[haroldadmin]

I think the job scheduling process used here could be greatly simplified using the select expression from the Kotlin Coroutines library.

The idea is to always process setState actions before getState actions, even when there are nested actions. A simple implementation could look like this:

typealias Reducer = S.() -> S
typealias Action = (S) -> Unit

val setStateJobs = Channel<Reducer<S>>(Channel.UNLIMITED)
val getStateJobs = Channel<Action<S>>(Channel.UNLIMITED)

coroutineScope.launch {
  while (isActive) {
    select<Unit> {
      setStateJobs.onReceive { reducer -> 
        val newState = reducer.invoke(currentState)
        updateState(newState)
      }
      getStateJobs.onReceive { action -> 
        action.invoke(currentState) 
      } 
    }
  }
}

The select expression lets us wait on multiple channels together. These channels are specified using different 'cases' in the select expression. It suspends while no items can be received in any of its cases, and resumes to execute the first received item. Therefore, it is safe and efficient to run it in an while loop scoped to the lifetime of the enclosing CoroutineScope.

The select expression is biased towards the first case (setStateJobs.onReceiver { ... }), and therefore it always processes the reducers before actions in this implementation.

Here is the complete implementation in my own library, and here are the tests for it.

Your thoughts, @gpeal @digitalbuddha @elihart @Dwite @ATizik ?

[rougsig]

ConflatedBroadcastChannel can drop events. It it expected behavior for MvCorStateStore?

  @Test
  fun test_ConflatedBroadcastChannel_runBlocking() {
    runBlocking {
      val channel = ConflatedBroadcastChannel<Int>()     
      async {
        channel.asFlow().collect {
          println("collected $it")
        }
      }
      async {
        (0 until 5).forEach {
          println("send $it")
          channel.send(it)
        }
        channel.cancel()
      }      
    }    
    // console out
    // send 0
    // send 1
    // send 2
    // send 3
    // send 4
    // collected 0
    // collected 4
  }

[gpeal]

@rougsig This PR has been superceded by #347

[gpeal]

@ATizik Thank you for the work you did to get MvRx started on a coroutines path. We have begin the work on 2.0 on the release/2.0.0 branch. This included #347 which makes this obsolete but thanks again for the time you put in!