Fix an issue where the effect flow wasn't shared.

flowbius/src/main/kotlin/com/trello/flowbius/FlowConnectables.kt

@@ -19,21 +19,20 @@ private class FlowConnectable<I, O>(
   private val mapper: FlowTransformer<I, O>
 ) : Connectable<I, O> {
   override fun connect(output: Consumer<O>): Connection<I> {
-    val channel = Channel<I>()
+    val sharedFlow = MutableSharedFlow<I>(replay = 1, extraBufferCapacity = Int.MAX_VALUE)
 
     val job = GlobalScope.launch(Dispatchers.Unconfined + context, start = CoroutineStart.ATOMIC) {
-      channel.receiveAsFlow()
+      sharedFlow
         .run { mapper(this) }
         .collect { output.accept(it) }
     }
 
     return object : Connection<I> {
       override fun accept(value: I) {
-        channel.trySendBlocking(value)
+        sharedFlow.tryEmit(value)
       }
 
       override fun dispose() {
-        channel.close()
         job.cancel()
       }
     }

flowbius/src/test/kotlin/com/trello/flowbius/FlowMobiusEffectRouterTest.kt

@@ -3,7 +3,6 @@ package com.trello.flowbius
 import app.cash.turbine.test
 import com.spotify.mobius.functions.Function
 import com.spotify.mobius.test.RecordingConsumer
-import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.collect
 import kotlinx.coroutines.flow.flowOf
 import kotlinx.coroutines.flow.map
@@ -19,7 +18,6 @@ class FlowMobiusEffectRouterTest {
   private lateinit var consumer: RecordingConsumer<TestEffect.C>
   private lateinit var action: TestAction
   private lateinit var router: FlowTransformer<TestEffect, TestEvent>
-  private lateinit var sharedFlow: MutableSharedFlow<TestEffect>
 
   @Before
   fun setup() {
@@ -33,7 +31,6 @@ class FlowMobiusEffectRouterTest {
       addFunction<TestEffect.E> { e -> TestEvent.E(e.id) }
       addFunction(Function<TestEffect.F, TestEvent> { value -> TestEvent.F(value.id) })
     }
-    sharedFlow = MutableSharedFlow(extraBufferCapacity = 1)
   }
 
   @Test

flowbius/src/test/kotlin/com/trello/flowbius/FlowMobiusLoopTest.kt

@@ -10,9 +10,7 @@ import com.spotify.mobius.test.RecordingConnection
 import com.spotify.mobius.test.RecordingConsumer
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.delay
-import kotlinx.coroutines.flow.MutableSharedFlow
-import kotlinx.coroutines.flow.flow
-import kotlinx.coroutines.flow.map
+import kotlinx.coroutines.flow.*
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
 import org.junit.Assert.assertEquals
@@ -168,4 +166,37 @@ class FlowMobiusLoopTest {
 
     job.join()
   }
+
+  /**
+   * [Dispatchers.Unconfined] has unfortunate nested behavior for us, which is that its order of execution is undefined.
+   * This can cause problems where start effects aren't ever handled because the start effect fires *before* the loop
+   * is fully configured. This test verifies that we won't miss start effects in this situation.
+   */
+  @Test
+  fun startEffectWorksWithUnconfinedDispatchersWithFlatMap() = runBlocking {
+
+    data class LoadEffect(val id: String)
+    data class LoadEvent(val id: String)
+
+    val job = launch(Dispatchers.Unconfined) {
+      val loop = FlowMobius.loop<String, LoadEvent, LoadEffect>(
+        update = { _, event -> next("Loaded data: ${event.id}") },
+        context = Dispatchers.Unconfined,
+        effectHandler = subtypeEffectHandler { addTransformer<LoadEffect> { s -> s.flatMapLatest { flowOf(LoadEvent(it.id)) } } }
+      )
+        .eventRunner(::ImmediateWorkRunner)
+        .effectRunner(::ImmediateWorkRunner)
+        .startFrom("No data loaded", setOf(LoadEffect("abc")))
+
+      // I hate to add this delay, but we have to give the loop a suspend point with which to initially process data.
+      //
+      // This implies that the start effect will not *immediately* fire off on loop startup, which is unfortunate
+      // but unavoidable given what happens with scheduling nested Dispatchers.Unconfined launches.
+      delay(10)
+
+      assertEquals("Loaded data: abc", loop.mostRecentModel)
+    }
+
+    job.join()
+  }
 }