Restore source compatibility with the Interactor lifecycle changes (StateFlow over SharedFlow).

Author: psteiger

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 == null || it < range.endInclusive }.collect()
+    takeWhile { 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

@@ -27,11 +27,12 @@ import io.reactivex.Observable
 import javax.inject.Inject
 import kotlin.properties.ReadWriteProperty
 import kotlin.reflect.KProperty
+import kotlinx.coroutines.ExperimentalForInheritanceCoroutinesApi
 import kotlinx.coroutines.channels.BufferOverflow
-import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.FlowCollector
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.MutableStateFlow
-import kotlinx.coroutines.flow.filterNotNull
+import kotlinx.coroutines.flow.SharedFlow
 import kotlinx.coroutines.rx2.asObservable
 
 /**
@@ -48,8 +49,7 @@ public abstract class Interactor<P : Any, R : Router<*>>() : InteractorType, Rib
   private val useStateFlow
     get() = RibEvents.useStateFlowInteractorEvent
 
-  @VisibleForTesting
-  internal val _lifecycleFlow: MutableSharedFlow<InteractorEvent?> =
+  private val _lifecycleFlow: MutableSharedFlow<InteractorEvent?> =
     if (useStateFlow) {
       MutableStateFlow(null)
     } else {
@@ -59,8 +59,7 @@ public abstract class Interactor<P : Any, R : Router<*>>() : InteractorType, Rib
         BufferOverflow.DROP_OLDEST,
       )
     }
-  public open val lifecycleFlow: Flow<InteractorEvent>
-    get() = _lifecycleFlow.filterNotNull()
+  public open val lifecycleFlow: SharedFlow<InteractorEvent> = NonNullSharedFlow(_lifecycleFlow)
 
   @Volatile private var _lifecycleObservable: Observable<InteractorEvent>? = null
 
@@ -86,16 +85,16 @@ 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 ---- //
 
   override fun isAttached(): Boolean =
-    _lifecycleFlow.replayCache.lastOrNull() == InteractorEvent.ACTIVE
+    lifecycleFlow.replayCache.lastOrNull() == InteractorEvent.ACTIVE
 
   override fun handleBackPress(): Boolean = false
 
@@ -233,3 +232,20 @@ public abstract class Interactor<P : Any, R : Router<*>>() : InteractorType, Rib
       }
   }
 }
+
+// See https://github.com/Kotlin/kotlinx.coroutines/issues/2514
+@OptIn(ExperimentalForInheritanceCoroutinesApi::class)
+private class NonNullSharedFlow<R : Any>(
+  private val upstream: SharedFlow<R?>,
+) : SharedFlow<R> {
+  override val replayCache: List<R>
+    get() = upstream.replayCache.filterNotNull()
+
+  override suspend fun collect(collector: FlowCollector<R>): Nothing {
+    upstream.collect { value ->
+      if (value != null) {
+        collector.emit(value)
+      }
+    }
+  }
+}

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

@@ -25,8 +25,6 @@ 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
@@ -229,13 +227,11 @@ class InteractorAndRouterTest {
     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)
   }
 
@@ -244,13 +240,11 @@ class InteractorAndRouterTest {
     // 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)
   }