Wrap subscription action with lifecycle whenStarted (#665)

* Wrap subscription action with lifecycle `whenStarted`

The specifics of behaviour of `flowWhenStarted` implementation implicitly used in resolve subscription will allow submitted action to resume continuation even if lifecycle is moved into stopped state. For example:

```
// Fragment

override fun onCreate(savedInstanceState: Bundle?) {
  viewModel.onEach(ViewState::isVisible) { isVisible ->
    ...
    binding.someview.reveal()  // (1) suspendable, performs animation
    binding.someview.text = "" // (2) touches other UI after continuation
    ...
  }
}
```

Users can observe crash in above example if lifecycle of view moved into stopped state while coroutine was suspended at (1) suspension point. The reason is that submitted action via `onEach` won't be paused / canceled when reached suspension point and lifecycle moved into stopped state (see implementations of `collectLatest` and `flowWhenStarted` together).

The recommendation was to wrap submitted `onEach` action into `whenStarted` on the user side to make coroutine paused when lifecycle is stopped.

But why not change this on mavericks side and address this issue at the core?

Author: sav007

mvrx/src/main/kotlin/com/airbnb/mvrx/FlowExtensions.kt

@@ -3,6 +3,7 @@ package com.airbnb.mvrx
 import androidx.lifecycle.DefaultLifecycleObserver
 import androidx.lifecycle.LifecycleOwner
 import androidx.lifecycle.lifecycleScope
+import androidx.lifecycle.whenStarted
 import kotlinx.coroutines.CoroutineStart
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.flow.Flow
@@ -41,7 +42,13 @@ internal fun <T : Any?> Flow<T>.collectLatest(
         // This is necessary when Dispatchers.Main.immediate is used in scope.
         // Coroutine is launched with start = CoroutineStart.UNDISPATCHED to perform dispatch only once.
         yield()
-        flow.collectLatest(action)
+        flow.collectLatest {
+            if (MavericksTestOverrides.FORCE_DISABLE_LIFECYCLE_AWARE_OBSERVER) {
+                action(it)
+            } else {
+                lifecycleOwner.whenStarted { action(it) }
+            }
+        }
     }
 }