[Android] Ensure main thread internally.

formula-android/src/main/java/com/instacart/formula/android/internal/ActivityManager.kt

@@ -6,6 +6,7 @@ import com.instacart.formula.android.events.ActivityResult
 import com.instacart.formula.android.FragmentEnvironment
 import com.instacart.formula.android.ActivityStore
 import com.instacart.formula.android.FormulaFragment
+import com.instacart.formula.android.FragmentFlowState
 import com.instacart.formula.android.ViewFactory
 import io.reactivex.rxjava3.disposables.CompositeDisposable
 import io.reactivex.rxjava3.disposables.Disposable
@@ -61,12 +62,13 @@ internal class ActivityManager<Activity : FragmentActivity>(
         delegate.attachActivity(activity)
         delegate.onLifecycleStateChanged(Lifecycle.State.CREATED)
         val renderView = fragmentRenderView ?: throw callOnPreCreateException(activity)
-        uiSubscription = fragmentState.subscribe {
-            Utils.assertMainThread()
 
+        val updateScheduler = AndroidUpdateScheduler<FragmentFlowState> {
             renderView.render(it)
             store.onRenderFragmentState?.invoke(activity, it)
         }
+
+        uiSubscription = fragmentState.subscribe(updateScheduler::emitUpdate)
     }
 
     fun onActivityStarted(activity: Activity) {

formula-android/src/main/java/com/instacart/formula/android/internal/ActivityStoreContextImpl.kt

@@ -71,9 +71,11 @@ internal class ActivityStoreContextImpl<Activity : FragmentActivity> : ActivityS
     }
 
     override fun send(effect: Activity.() -> Unit) {
-        // We allow emitting effects only after activity has started
-        startedActivity()?.effect() ?: run {
-            // Log missing activity.
+        Utils.executeOnMainThread {
+            // We allow emitting effects only after activity has started
+            startedActivity()?.effect() ?: run {
+                // Log missing activity.
+            }
         }
     }
 

formula-android/src/main/java/com/instacart/formula/android/internal/AndroidUpdateScheduler.kt

@@ -0,0 +1,71 @@
+package com.instacart.formula.android.internal
+
+import java.util.concurrent.atomic.AtomicBoolean
+import java.util.concurrent.atomic.AtomicReference
+
+/**
+ * Handles state update scheduling to the main thread. If update arrives on a background thread,
+ * it will added it the main thread queue. It will throw away a pending update if a new update
+ * arrives.
+ */
+class AndroidUpdateScheduler<Value : Any>(
+    private val update: (Value) -> Unit,
+) {
+    /**
+     * If not null, that means that we have an update pending.
+     */
+    private val pendingValue = AtomicReference<Value>()
+
+    /**
+     * Defines if an update is currently scheduled.
+     */
+    private val updateScheduled = AtomicBoolean(false)
+
+    /**
+     * To avoid re-entry, we track if [updateRunnable] is currently handling an update.
+     */
+    private var isUpdating = false
+
+    private val updateRunnable = object :Runnable {
+        override fun run() {
+            updateScheduled.set(false)
+
+            var localPending = pendingValue.getAndSet(null)
+            while (localPending != null) {
+                // Handle the update
+                isUpdating = true
+                update(localPending)
+                isUpdating = false
+
+                // Check if another update arrived while we were processing.
+                localPending =  pendingValue.getAndSet(null)
+
+                if (localPending != null) {
+                    // We will take over processing, so let's clear the message
+                    Utils.mainThreadHandler.removeCallbacks(this)
+                }
+            }
+        }
+    }
+
+    fun emitUpdate(value: Value) {
+        // Set pending value
+        pendingValue.set(value)
+
+        if (Utils.isMainThread()) {
+            if (isUpdating) {
+                // Let's exit and let the [updateRunnable] to pick up the change
+                return
+            } else {
+                // Since we are on main thread, let's force run it
+                updateRunnable.run()
+            }
+        } else {
+            // If no update is scheduled, schedule one
+            if (updateScheduled.compareAndSet(false, true)) {
+
+                Utils.mainThreadHandler.post(updateRunnable)
+            }
+        }
+    }
+}

formula-android/src/main/java/com/instacart/formula/android/internal/CompositeBinding.kt

@@ -35,10 +35,10 @@ internal class CompositeBinding<ParentComponent, ScopedComponent>(
             val component = state.component
             if (component != null) {
                 val childInput = Input(
-                    input.environment,
-                    component.component,
-                    input.activeFragments,
-                    input.onInitializeFeature,
+                    environment = input.environment,
+                    component = component.component,
+                    activeFragments = input.activeFragments,
+                    onInitializeFeature = input.onInitializeFeature,
                 )
                 bindings.forEachIndices {
                     it.bind(context, childInput)

formula-android/src/main/java/com/instacart/formula/android/internal/FeatureBinding.kt

@@ -33,6 +33,7 @@ internal class FeatureBinding<in Component, in Dependencies, in Key : FragmentKe
                         if (binds(key)) {
                             Action.onData(fragmentId).onEvent {
                                 transition {
+                                    // TODO: should this happen on the main thread? It needs to be available to main thread
                                     try {
                                         val dependencies = toDependencies(input.component)
                                         val feature = input.environment.fragmentDelegate.initializeFeature(

formula-android/src/main/java/com/instacart/formula/android/internal/FragmentFlowRenderView.kt

@@ -42,6 +42,7 @@ internal class FragmentFlowRenderView(
 
     private val featureProvider = object : FeatureProvider {
         override fun getFeature(id: FragmentId): FeatureEvent? {
+            // TODO: should we initialize feature if it's missing
             return fragmentState?.features?.get(id)
         }
     }
@@ -118,6 +119,8 @@ internal class FragmentFlowRenderView(
     }
 
     fun render(state: FragmentFlowState) {
+        Utils.assertMainThread()
+
         fragmentState = state
         updateVisibleFragments(state)
     }

formula-android/src/main/java/com/instacart/formula/android/internal/StreamConfiguratorIml.kt

@@ -18,14 +18,17 @@ internal class StreamConfiguratorIml<out Activity : FragmentActivity>(
         val stateEmissions = Observable.combineLatest(
             state,
             context.activityStartedEvents(),
-            BiFunction<State, Unit, State> { state, event ->
-                state
+            BiFunction { stateValue, _ ->
+                stateValue
             }
         )
-        return stateEmissions.subscribe { state ->
+
+        val updateScheduler = AndroidUpdateScheduler<State> { stateValue ->
             context.startedActivity()?.let {
-                update(it, state)
+                update(it, stateValue)
             }
         }
+
+        return stateEmissions.subscribe(updateScheduler::emitUpdate)
     }
 }

formula-android/src/main/java/com/instacart/formula/android/internal/Utils.kt

@@ -1,11 +1,26 @@
 package com.instacart.formula.android.internal
 
+import android.os.Handler
 import android.os.Looper
 
 internal object Utils {
+    internal val mainThreadHandler = Handler(Looper.getMainLooper())
+
     fun assertMainThread() {
-        if (Looper.getMainLooper() != Looper.myLooper()) {
+        if (!isMainThread()) {
             throw IllegalStateException("should be called on main thread: ${Thread.currentThread()}")
         }
     }
+
+    inline fun executeOnMainThread(crossinline runnable: () -> Unit) {
+        if (isMainThread()) {
+            runnable()
+        } else {
+            mainThreadHandler.post { runnable() }
+        }
+    }
+
+    fun isMainThread(): Boolean {
+        return Looper.getMainLooper() == Looper.myLooper()
+    }
 }