diff --git a/CHANGELOG.md b/CHANGELOG.md index e18a9eff..95c5091d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,28 @@ #### Additions +([#179](https://github.com/badoo/MVICore/pull/179)): +Introduced the ability to specify observation scheduler within the `Binder` class (see the `observeOn` DSL below), as well as the `observeOn` infix operator for `Connection` class (related to `Binder`). + +An example of how you might use this is as follows: + +```kotlin +testLifecycleOwner.lifecycle.createDestroy { + observeOn(mainScheduler) { + bind(events to uiConsumer1) + bind(events to uiConsumer2) + } + observeOn(backgroundScheduler) { + bind(events to backgroundConsumer1) + bind(events to backgroundConsumer2) + } + bind(events to uiConsumer3 observeOn mainScheduler) + bind(events to backgroundConsumer3 observeOn backgroundScheduler) +} +``` + +See more details in [advanced binder](../binder/binder-advanced/#setting-connections-observation-scheduler) section. + ([#177](https://github.com/badoo/MVICore/pull/177)): Updated AndroidX appcompat to 1.4.1 and lifecycle to 2.5.1. Also updated Compile and Target SDK to API 33. diff --git a/binder/src/main/java/com/badoo/binder/Binder.kt b/binder/src/main/java/com/badoo/binder/Binder.kt index e98ce979..f21114ad 100644 --- a/binder/src/main/java/com/badoo/binder/Binder.kt +++ b/binder/src/main/java/com/badoo/binder/Binder.kt @@ -8,13 +8,14 @@ import com.badoo.binder.middleware.wrapWithMiddleware import io.reactivex.Observable import io.reactivex.Observable.wrap import io.reactivex.ObservableSource +import io.reactivex.Scheduler import io.reactivex.disposables.CompositeDisposable import io.reactivex.disposables.Disposable import io.reactivex.functions.Consumer import io.reactivex.rxkotlin.plusAssign class Binder( - private val lifecycle: Lifecycle? = null + private val lifecycle: Lifecycle? = null, ) : Disposable { private val disposables = CompositeDisposable() private val connections = mutableListOf, Middleware<*, *>?>>() @@ -85,12 +86,14 @@ class Binder( middleware.onBind(connection) this + .optionalObserveOn(connection.scheduler) .doOnNext { middleware.onElement(connection, it) } .doFinally { middleware.onComplete(connection) } .subscribe(middleware) } else { - subscribe(connection.to) + optionalObserveOn(connection.scheduler) + .subscribe(connection.to) } } @@ -142,4 +145,36 @@ class Binder( fun clear() { disposables.clear() } + + private fun Observable.optionalObserveOn(scheduler: Scheduler?) = + if (scheduler != null) { + observeOn(scheduler) + } else { + this + } + + fun observeOn(scheduler: Scheduler, f: BinderSchedulerWrapper.() -> Unit) { + BinderSchedulerWrapper(this, scheduler).apply(f) + } + + class BinderSchedulerWrapper(private val binder: Binder, private val scheduler: Scheduler?) { + fun bind(connection: Pair, Consumer>) { + binder.bind( + Connection( + from = connection.first, + to = connection.second, + connector = null, + scheduler = scheduler + ) + ) + } + + fun bind(connection: Connection) { + if (scheduler != null) { + binder.bind(connection.observeOn(scheduler)) + } else { + binder.bind(connection) + } + } + } } diff --git a/binder/src/main/java/com/badoo/binder/Connection.kt b/binder/src/main/java/com/badoo/binder/Connection.kt index 8fa8b0ac..e14a1a2d 100644 --- a/binder/src/main/java/com/badoo/binder/Connection.kt +++ b/binder/src/main/java/com/badoo/binder/Connection.kt @@ -3,13 +3,15 @@ package com.badoo.binder import com.badoo.binder.connector.Connector import com.badoo.binder.connector.NotNullConnector import io.reactivex.ObservableSource +import io.reactivex.Scheduler import io.reactivex.functions.Consumer data class Connection( val from: ObservableSource? = null, val to: Consumer, val connector: Connector? = null, - val name: String? = null + val name: String? = null, + val scheduler: Scheduler? = null, ) { companion object { private const val ANONYMOUS: String = "anonymous" @@ -39,7 +41,19 @@ infix fun Pair, Consumer>.named(name: String): name = name ) +infix fun Pair, Consumer>.observeOn(scheduler: Scheduler): Connection = + Connection( + from = first, + to = second, + scheduler = scheduler + ) + infix fun Connection.named(name: String) = copy( name = name ) + +infix fun Connection.observeOn(scheduler: Scheduler) = + copy( + scheduler = scheduler + ) diff --git a/documentation/binder/binder-advanced.md b/documentation/binder/binder-advanced.md index 9a278fe4..687329bc 100644 --- a/documentation/binder/binder-advanced.md +++ b/documentation/binder/binder-advanced.md @@ -37,7 +37,7 @@ binder.bind(output to input using OutputToInput) ## Naming connections -And you can optionally give names to any connection: +You can optionally give names to any connection: ```kotlin binder.bind(input to output named "MyConnection") // or @@ -49,3 +49,20 @@ Naming a connection signals that it's important to you. This will make more sens - You'll see connections with their respective names in the time-travel debug menu - You'll see connection names in logs if you use LoggingMiddleware - You can opt to dynamically add `Middlewares` only to named connections (if that's what you want) + +## Setting connections observation scheduler + +You can optionally set the observation scheduler for any connection: +```kotlin +binder.bind(input to output observeOn scheduler) +``` + +You can also use `Binder.observeOn` to reduce repetition: +```kotlin +binder.observeOn(scheduler) { + bind(input1 to output1) + bind(input2 to output2) +} +``` + +Specifying an observation scheduler ensures that the output is called on the specified scheduler. diff --git a/mvicore-android/src/test/java/com/badoo/mvicore/android/lifecycle/LifecycleExtensionsTest.kt b/mvicore-android/src/test/java/com/badoo/mvicore/android/lifecycle/LifecycleExtensionsTest.kt index 805ee164..2cfa0112 100644 --- a/mvicore-android/src/test/java/com/badoo/mvicore/android/lifecycle/LifecycleExtensionsTest.kt +++ b/mvicore-android/src/test/java/com/badoo/mvicore/android/lifecycle/LifecycleExtensionsTest.kt @@ -3,11 +3,15 @@ package com.badoo.mvicore.android.lifecycle import androidx.arch.core.executor.ArchTaskExecutor import androidx.arch.core.executor.TaskExecutor import androidx.lifecycle.Lifecycle +import com.badoo.binder.observeOn import io.reactivex.functions.Consumer +import io.reactivex.internal.schedulers.RxThreadFactory +import io.reactivex.plugins.RxJavaPlugins import io.reactivex.subjects.PublishSubject import org.junit.After import org.junit.Before import org.junit.Test +import java.util.concurrent.CountDownLatch import kotlin.test.assertEquals class LifecycleExtensionsTest { @@ -15,6 +19,14 @@ class LifecycleExtensionsTest { private val consumerTester = ConsumerTester() private val testLifecycleOwner = TestLifecycleOwner() + private val mainScheduler = RxJavaPlugins + .createSingleScheduler(RxThreadFactory("main", Thread.NORM_PRIORITY, false)) + .apply { start() } + + private val backgroundScheduler = RxJavaPlugins + .createSingleScheduler(RxThreadFactory("background", Thread.NORM_PRIORITY, false)) + .apply { start() } + @Before fun setup() { ArchTaskExecutor.getInstance() @@ -225,11 +237,89 @@ class LifecycleExtensionsTest { assertEquals(false, subject.hasObservers()) } + @Test + fun `GIVEN initial lifecycle is created AND createDestroy with observe on schedulers dsl WHEN event emitted THEN verify correct thread called`() { + val testThreadName = Thread.currentThread().name + + val subject = PublishSubject.create() + val mainThreadConsumerTester = ConsumerTester() + val backgroundThreadConsumerTester = ConsumerTester() + val unconfinedThreadConsumerTester = ConsumerTester() + + val countDownLatch = CountDownLatch(3) + + testLifecycleOwner.lifecycle.createDestroy { + observeOn(mainScheduler) { + bind(subject to Consumer { + mainThreadConsumerTester.accept(Unit) + countDownLatch.countDown() + }) + } + observeOn(backgroundScheduler) { + bind(subject to Consumer { + backgroundThreadConsumerTester.accept(Unit) + countDownLatch.countDown() + }) + } + bind(subject to Consumer { + unconfinedThreadConsumerTester.accept(Unit) + countDownLatch.countDown() + }) + } + testLifecycleOwner.state = Lifecycle.State.CREATED + + subject.onNext(Unit) + + countDownLatch.await() + + mainThreadConsumerTester.verifyThreadName("main") + backgroundThreadConsumerTester.verifyThreadName("background") + unconfinedThreadConsumerTester.verifyThreadName(testThreadName) + } + + @Test + fun `GIVEN initial lifecycle is created AND createDestroy with schedulers infix WHEN event emitted THEN verify correct thread called`() { + val testThreadName = Thread.currentThread().name + + val subject = PublishSubject.create() + val mainThreadConsumerTester = ConsumerTester() + val backgroundThreadConsumerTester = ConsumerTester() + val unconfinedThreadConsumerTester = ConsumerTester() + + val countDownLatch = CountDownLatch(3) + + testLifecycleOwner.lifecycle.createDestroy { + bind(subject to Consumer { + mainThreadConsumerTester.accept(Unit) + countDownLatch.countDown() + } observeOn mainScheduler) + bind(subject to Consumer { + backgroundThreadConsumerTester.accept(Unit) + countDownLatch.countDown() + } observeOn backgroundScheduler) + bind(subject to Consumer { + unconfinedThreadConsumerTester.accept(Unit) + countDownLatch.countDown() + }) + } + testLifecycleOwner.state = Lifecycle.State.CREATED + + subject.onNext(Unit) + + countDownLatch.await() + + mainThreadConsumerTester.verifyThreadName("main") + backgroundThreadConsumerTester.verifyThreadName("background") + unconfinedThreadConsumerTester.verifyThreadName(testThreadName) + } + private class ConsumerTester : Consumer { private var wasCalled: Boolean = false + lateinit var threadName: String override fun accept(t: Unit?) { wasCalled = true + threadName = Thread.currentThread().name } fun verifyInvoked() { @@ -239,5 +329,9 @@ class LifecycleExtensionsTest { fun verifyNotInvoked() { assertEquals(false, wasCalled) } + + fun verifyThreadName(name: String) { + assertEquals(true, threadName.startsWith(name)) + } } }