Use a direct dispatcher in Flow -> Rx conversions.

A direct dispatcher mimics the Rx always synchronous-by-default behavior (unless `subscribeOn`/`observeOn` is used), which is different to the `Dispatchers.Unconfined` behavior (default used by `Flow.asObservable`) which may form event loops for nested coroutines (asynchronous).

Author: psteiger

android/conventions/src/main/kotlin/ribs.spotless-convention.gradle.kts

@@ -26,6 +26,7 @@ configure<com.diffplug.gradle.spotless.SpotlessExtension> {
     endWithNewline()
   }
   kotlin {
+    toggleOffOn()
     target("**/*.kt")
     ktlint(libs.versions.ktlint.get()).editorConfigOverride(
         mapOf(

android/libraries/rib-android/src/main/kotlin/com/uber/rib/core/RibActivity.kt

@@ -60,7 +60,7 @@ abstract class RibActivity :
 
   @Volatile private var _lifecycleObservable: Observable<ActivityLifecycleEvent>? = null
   private val lifecycleObservable
-    get() = ::_lifecycleObservable.setIfNullAndGet { lifecycleFlow.asObservable() }
+    get() = ::_lifecycleObservable.setIfNullAndGet { lifecycleFlow.asObservable(DirectDispatcher) }
 
   private val _callbacksFlow =
     MutableSharedFlow<ActivityCallbackEvent>(0, 1, BufferOverflow.DROP_OLDEST)
@@ -69,7 +69,7 @@ abstract class RibActivity :
 
   @Volatile private var _callbacksObservable: Observable<ActivityCallbackEvent>? = null
   private val callbacksObservable
-    get() = ::_callbacksObservable.setIfNullAndGet { callbacksFlow.asObservable() }
+    get() = ::_callbacksObservable.setIfNullAndGet { callbacksFlow.asObservable(DirectDispatcher) }
 
   /** @return an observable of this activity's lifecycle events. */
   final override fun lifecycle(): Observable<ActivityLifecycleEvent> = lifecycleObservable

android/libraries/rib-base/src/main/kotlin/com/uber/rib/core/FlowAsScope.kt

@@ -14,6 +14,7 @@
  * limitations under the License.
  */
 @file:JvmSynthetic
+@file:Suppress("invisible_reference", "invisible_member")
 
 package com.uber.rib.core
 
@@ -41,7 +42,7 @@ internal fun <T : Comparable<T>> SharedFlow<T>.asScopeCompletable(
   context: CoroutineContext = EmptyCoroutineContext,
 ): CompletableSource {
   ensureAlive(range)
-  return rxCompletable(RibDispatchers.Unconfined + context) {
+  return rxCompletable(DirectDispatcher + context) {
     takeWhile { it < range.endInclusive }.collect()
   }
 }

android/libraries/rib-base/src/main/kotlin/com/uber/rib/core/Interactor.kt

@@ -13,6 +13,8 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+@file:Suppress("invisible_reference", "invisible_member")
+
 package com.uber.rib.core
 
 import androidx.annotation.CallSuper
@@ -46,7 +48,7 @@ public abstract class Interactor<P : Any, R : Router<*>>() : InteractorType, Rib
 
   @Volatile private var _lifecycleObservable: Observable<InteractorEvent>? = null
   private val lifecycleObservable
-    get() = ::_lifecycleObservable.setIfNullAndGet { lifecycleFlow.asObservable() }
+    get() = ::_lifecycleObservable.setIfNullAndGet { lifecycleFlow.asObservable(DirectDispatcher) }
 
   private val routerDelegate = InitOnceProperty<R>()
 

android/libraries/rib-base/src/main/kotlin/com/uber/rib/core/Presenter.kt

@@ -40,7 +40,7 @@ public abstract class Presenter : ScopeProvider, RibActionEmitter {
 
   @Volatile private var _lifecycleObservable: Observable<PresenterEvent>? = null
   private val lifecycleObservable
-    get() = ::_lifecycleObservable.setIfNullAndGet { lifecycleFlow.asObservable() }
+    get() = ::_lifecycleObservable.setIfNullAndGet { lifecycleFlow.asObservable(DirectDispatcher) }
 
   /** @return `true` if the presenter is loaded, `false` if not. */
   protected var isLoaded: Boolean = false

android/libraries/rib-base/src/test/kotlin/com/uber/rib/core/InteractorTest.kt

@@ -0,0 +1,35 @@
+/*
+ * Copyright (C) 2023. Uber Technologies
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.uber.rib.core
+
+import com.uber.rib.core.lifecycle.InteractorEvent
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.test.runTest
+import org.junit.Test
+
+@OptIn(ExperimentalCoroutinesApi::class)
+class InteractorTest {
+  @Test
+  fun assertInteractorEventEagerness() = runTest {
+    assertLifecycleEventEagerness(InteractorEvent.ACTIVE) {
+      val interactor = object : Interactor<Any, Router<*>>() {}
+      interactor.apply {
+        setPresenter(Unit)
+        dispatchAttach(null)
+      }
+    }
+  }
+}

android/libraries/rib-base/src/test/kotlin/com/uber/rib/core/LifecycleEventEagerness.kt

@@ -0,0 +1,50 @@
+/*
+ * Copyright (C) 2023. Uber Technologies
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.uber.rib.core
+
+import com.google.common.truth.Truth.assertThat
+import com.uber.autodispose.lifecycle.LifecycleScopeProvider
+import io.reactivex.disposables.Disposable
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.test.TestScope
+import kotlinx.coroutines.test.advanceUntilIdle
+
+@OptIn(ExperimentalCoroutinesApi::class)
+internal inline fun <T> TestScope.assertLifecycleEventEagerness(
+  event: T,
+  crossinline lifecycleScopeProviderProducer: () -> LifecycleScopeProvider<T>,
+) {
+  val events = mutableListOf<Event<T>>()
+  var disposable: Disposable? = null
+  launch(Dispatchers.Unconfined) {
+    disposable =
+      lifecycleScopeProviderProducer().lifecycle().subscribe { lifecycleEvent ->
+        events.add(Event.Lifecycle(lifecycleEvent))
+      }
+    events.add(Event.AfterSubscription)
+  }
+  advanceUntilIdle()
+  assertThat(events).isEqualTo(listOf(Event.Lifecycle(event), Event.AfterSubscription))
+  disposable?.dispose()
+}
+
+@PublishedApi
+internal sealed interface Event<out T> {
+  data class Lifecycle<T>(val event: T) : Event<T>
+  object AfterSubscription : Event<Nothing>
+}

android/libraries/rib-coroutines/src/main/kotlin/com/uber/rib/core/DirectDispatcher.kt

@@ -0,0 +1,39 @@
+/*
+ * Copyright (C) 2023. Uber Technologies
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.uber.rib.core
+
+import kotlin.coroutines.CoroutineContext
+import kotlinx.coroutines.CoroutineDispatcher
+import kotlinx.coroutines.Runnable
+
+/**
+ * A dispatcher that immediately executes the [Runnable] on the same stack frame, without
+ * potentially forming event loops like [Unconfined][kotlinx.coroutines.Dispatchers.Unconfined] or
+ * [Main.immediate][kotlinx.coroutines.MainCoroutineDispatcher.immediate] in case of nested
+ * coroutines.
+ *
+ * For more context, see the following issues on `kotlinx.coroutines` GitHub repository:
+ * <!-- spotless:off -->
+ * 1. [Immediate dispatchers can cause spooky action at a distance](https://github.com/Kotlin/kotlinx.coroutines/issues/3760)
+ * 2. [Chaining rxSingle calls that use Unconfined dispatcher and blockingGet results in deadlock #3458](https://github.com/Kotlin/kotlinx.coroutines/issues/3458)
+ * 3. [Coroutines/Flow vs add/remove listener (synchronous execution) #3506](https://github.com/Kotlin/kotlinx.coroutines/issues/3506)
+ * <!-- spotless:on -->
+ */
+internal object DirectDispatcher : CoroutineDispatcher() {
+  override fun dispatch(context: CoroutineContext, block: Runnable) {
+    block.run()
+  }
+}

android/libraries/rib-router-navigator/src/main/kotlin/com/uber/rib/core/RouterNavigatorEvents.kt

@@ -13,6 +13,8 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+@file:Suppress("invisible_reference", "invisible_member")
+
 package com.uber.rib.core
 
 import io.reactivex.Observable
@@ -26,7 +28,7 @@ public class RouterNavigatorEvents private constructor() {
   private val _events = MutableSharedFlow<RouterNavigatorEvent>(0, 1, BufferOverflow.DROP_OLDEST)
 
   /** @return the stream which can be subcribed to listen for [RouterNavigatorEvent] */
-  public val events: Observable<RouterNavigatorEvent> = _events.asObservable()
+  public val events: Observable<RouterNavigatorEvent> = _events.asObservable(DirectDispatcher)
 
   @JvmSynthetic // Hide from Java consumers. In Java, `getEvents` resolves to the `events` property.
   @JvmName("_getEvents")

android/libraries/rib-screen-stack-base/build.gradle

@@ -27,6 +27,7 @@ dependencies {
     api(libs.rxjava2)
     api(libs.rxrelay2)
     api(libs.rxbinding)
+    implementation(project(":libraries:rib-coroutines"))
     implementation(libs.annotation)
     implementation(libs.autodispose.coroutines)
     implementation(libs.coroutines.android)