For Interactor, replace the shared flow with the stateflow to improve the ANR stacktraces and reduce ANRs

Author: rysh88

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

@@ -38,17 +38,17 @@ import kotlinx.coroutines.rx2.rxCompletable
  *    [range].
  */
 @CoreFriendModuleApi
-public fun <T : Comparable<T>> SharedFlow<T>.asScopeCompletable(
+public fun <T : Comparable<T>> SharedFlow<T?>.asScopeCompletable(
   range: ClosedRange<T>,
   context: CoroutineContext = EmptyCoroutineContext,
 ): CompletableSource {
   ensureAlive(range)
   return rxCompletable(RibDispatchers.Unconfined + context) {
-    takeWhile { it < range.endInclusive }.collect()
+    takeWhile { it == null || it < range.endInclusive }.collect()
   }
 }
 
-private fun <T : Comparable<T>> SharedFlow<T>.ensureAlive(range: ClosedRange<T>) {
+private fun <T : Comparable<T>> SharedFlow<T?>.ensureAlive(range: ClosedRange<T>) {
   val lastEmitted = replayCache.lastOrNull()
   when {
     lastEmitted == null || lastEmitted < range.start -> throw LifecycleNotStartedException()

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

@@ -28,8 +28,10 @@ import javax.inject.Inject
 import kotlin.properties.ReadWriteProperty
 import kotlin.reflect.KProperty
 import kotlinx.coroutines.channels.BufferOverflow
+import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableSharedFlow
-import kotlinx.coroutines.flow.SharedFlow
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.filterNotNull
 import kotlinx.coroutines.rx2.asObservable
 
 /**
@@ -42,9 +44,23 @@ public abstract class Interactor<P : Any, R : Router<*>>() : InteractorType, Rib
   @Inject public lateinit var injectedPresenter: P
 
   @CoreFriendModuleApi public var actualPresenter: P? = null
-  private val _lifecycleFlow = MutableSharedFlow<InteractorEvent>(1, 0, BufferOverflow.DROP_OLDEST)
-  public open val lifecycleFlow: SharedFlow<InteractorEvent>
-    get() = _lifecycleFlow
+
+  private val useStateFlow
+    get() = RibEvents.useStateFlowInteractorEvent
+
+  @VisibleForTesting
+  internal val _lifecycleFlow: MutableSharedFlow<InteractorEvent?> =
+    if (useStateFlow) {
+      MutableStateFlow(null)
+    } else {
+      MutableSharedFlow(
+        1,
+        0,
+        BufferOverflow.DROP_OLDEST,
+      )
+    }
+  public open val lifecycleFlow: Flow<InteractorEvent>
+    get() = _lifecycleFlow.filterNotNull()
 
   @Volatile private var _lifecycleObservable: Observable<InteractorEvent>? = null
 
@@ -70,11 +86,11 @@ public abstract class Interactor<P : Any, R : Router<*>>() : InteractorType, Rib
   final override fun correspondingEvents(): CorrespondingEventsFunction<InteractorEvent> =
     LIFECYCLE_MAP_FUNCTION
 
-  final override fun peekLifecycle(): InteractorEvent? = lifecycleFlow.replayCache.lastOrNull()
+  final override fun peekLifecycle(): InteractorEvent? = _lifecycleFlow.replayCache.lastOrNull()
 
   @OptIn(CoreFriendModuleApi::class)
   final override fun requestScope(): CompletableSource =
-    lifecycleFlow.asScopeCompletable(lifecycleRange)
+    _lifecycleFlow.asScopeCompletable(lifecycleRange)
 
   // ---- InteractorType overrides ---- //
 

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

@@ -35,6 +35,9 @@ public object RibEvents {
   @JvmStatic
   public val ribActionEvents: Observable<RibActionInfo> = mutableRibDurationEvents.asObservable()
 
+  internal var useStateFlowInteractorEvent: Boolean = false
+    private set
+
   /** Indicates if [ribActionEvents] will be emitting. */
   public var areRibActionEmissionsAllowed: Boolean = false
     @VisibleForTesting internal set
@@ -48,6 +51,12 @@ public object RibEvents {
     this.areRibActionEmissionsAllowed = true
   }
 
+  /** If true, the [Interactor] will use [MutableStateFlow] for the interactor events. */
+  @JvmStatic
+  public fun useStateFlowInteractorEvent() {
+    this.useStateFlowInteractorEvent = true
+  }
+
   /**
    * @param eventType [RibEventType]
    * @param child [Router]

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

@@ -31,7 +31,7 @@ import kotlinx.coroutines.CoroutineDispatcher
 import kotlinx.coroutines.CoroutineStart
 import kotlinx.coroutines.DelicateCoroutinesApi
 import kotlinx.coroutines.GlobalScope
-import kotlinx.coroutines.flow.SharedFlow
+import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.onCompletion
 import kotlinx.coroutines.flow.takeWhile
 import kotlinx.coroutines.launch
@@ -264,7 +264,7 @@ private fun getJobCoroutineContext(
 }
 
 private fun <T : Comparable<T>> Worker.bind(
-  lifecycle: SharedFlow<T>,
+  lifecycle: Flow<T>,
   lifecycleRange: ClosedRange<T>,
 ): WorkerUnbinder {
   val dispatcherAtBinder = RibCoroutinesConfig.deprecatedWorkerDispatcher

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

@@ -18,11 +18,18 @@ package com.uber.rib.core
 import com.google.common.truth.Truth
 import com.google.common.truth.Truth.assertThat
 import com.uber.autodispose.lifecycle.LifecycleEndedException
+import com.uber.autodispose.lifecycle.LifecycleNotStartedException
 import com.uber.rib.core.RibEvents.ribActionEvents
 import com.uber.rib.core.RibEventsUtils.assertRibActionInfo
 import com.uber.rib.core.RibRefWatcher.Companion.getInstance
 import com.uber.rib.core.lifecycle.InteractorEvent
+import io.reactivex.Completable
 import io.reactivex.observers.TestObserver
+import kotlinx.coroutines.flow.MutableSharedFlow
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.test.runTest
+import kotlinx.coroutines.withTimeoutOrNull
+import org.junit.Assert.fail
 import org.junit.Before
 import org.junit.Test
 import org.mockito.kotlin.any
@@ -216,6 +223,128 @@ class InteractorAndRouterTest {
     verify(ribRefWatcher, times(2)).watchDeletedObject(any())
   }
 
+  @Test
+  fun init_useStateFlow_eventIsNotFired() = runTest {
+    // Given
+    setupInteractorForStateFlow()
+    var eventCount = 0
+    val lifecycleFlow = interactor.lifecycleFlow
+    val _lifecycleFlow = interactor._lifecycleFlow
+
+    // When
+    withTimeoutOrNull(10) { lifecycleFlow.collect { eventCount++ } }
+
+    // Then
+    assertThat(_lifecycleFlow).isInstanceOf(MutableStateFlow::class.java)
+    assertThat(eventCount).isEqualTo(0)
+  }
+
+  @Test
+  fun init_eventIsNotFired() = runTest {
+    // When
+    var eventCount = 0
+    val lifecycleFlow = interactor.lifecycleFlow
+    val _lifecycleFlow = interactor._lifecycleFlow
+
+    // When
+    withTimeoutOrNull(10) { lifecycleFlow.collect { eventCount++ } }
+
+    // Then
+    assertThat(_lifecycleFlow).isInstanceOf(MutableSharedFlow::class.java)
+    assertThat(eventCount).isEqualTo(0)
+  }
+
+  @Test
+  fun requestScope_initAndUseStateFlow_throwsAnError() {
+    // Given
+    setupInteractorForStateFlow()
+
+    // When
+    try {
+      interactor.requestScope()
+      fail("Expected LifecycleEndedException")
+    } catch (e: LifecycleNotStartedException) {
+      // Then
+      // Expected exception, do nothing
+    }
+  }
+
+  @Test
+  fun requestScope_initWith_throwsAnError() {
+    // When
+    try {
+      interactor.requestScope()
+      fail("Expected LifecycleEndedException")
+    } catch (e: LifecycleNotStartedException) {
+      // Then
+      // Expected exception, do nothing
+    }
+  }
+
+  @Test
+  fun requestScope_useStateFlowAndAttached_requestScopeIsNotCompleted() {
+    // Given
+    setupInteractorForStateFlow()
+    interactor.dispatchAttach(null)
+
+    // When
+    Completable.wrap(interactor.requestScope())
+      .test()
+      // Then
+      .assertNotComplete()
+  }
+
+  @Test
+  fun requestScope_attached_requestScopeIsNotCompleted() {
+    // Given
+    interactor.dispatchAttach(null)
+
+    // When
+    Completable.wrap(interactor.requestScope())
+      .test()
+      // Then
+      .assertNotComplete()
+  }
+
+  @Test
+  fun requestScope_useStateFlowAndDetached_requestScopeIsCompleted() {
+    // Given
+    setupInteractorForStateFlow()
+    interactor.dispatchAttach(null)
+    val requestScope = Completable.wrap(interactor.requestScope())
+    interactor.dispatchDetach()
+
+    // When
+    requestScope
+      .test()
+      // Then
+      .assertComplete()
+  }
+
+  @Test
+  fun requestScope_detached_requestScopeIsCompleted() {
+    // Given
+    interactor.dispatchAttach(null)
+    val requestScope = Completable.wrap(interactor.requestScope())
+    interactor.dispatchDetach()
+
+    // When
+    requestScope
+      .test()
+      // Then
+      .assertComplete()
+  }
+
+  private fun setupInteractorForStateFlow() {
+    RibEvents.useStateFlowInteractorEvent()
+    val presenter: TestPresenter = mock()
+    val component: InteractorComponent<TestPresenter, TestInteractor> = mock {
+      on { presenter() } doReturn (presenter)
+    }
+    interactor = TestInteractor(childInteractor)
+    router = TestRouter(interactor, component)
+  }
+
   private fun addTwoNestedChildInteractors(): Router<TestInteractorB> {
     val component: InteractorComponent<TestPresenter, TestInteractorB> =
       object : InteractorComponent<TestPresenter, TestInteractorB> {