Remove RxJava2 dependency from Workflow Android UI integration.

Closes #1123.

kotlin/buildSrc/src/main/java/Dependencies.kt

@@ -22,6 +22,7 @@ object Dependencies {
     const val gridlayout = "androidx.gridlayout:gridlayout:1.0.0"
 
     object Lifecycle {
+      const val ktx = "androidx.lifecycle:lifecycle-runtime-ktx:2.2.0"
       const val reactivestreams = "androidx.lifecycle:lifecycle-reactivestreams-ktx:2.2.0"
     }
 

kotlin/samples/tictactoe/app/build.gradle.kts

@@ -37,6 +37,7 @@ dependencies {
   implementation(project(":workflow-tracing"))
 
   implementation(Dependencies.AndroidX.constraint_layout)
+  implementation(Dependencies.AndroidX.Lifecycle.ktx)
   implementation(Dependencies.okio)
   implementation(Dependencies.rxandroid2)
   implementation(Dependencies.Test.AndroidX.Espresso.idlingResource)

kotlin/samples/tictactoe/app/src/main/java/com/squareup/sample/mainactivity/MainActivity.kt

@@ -17,6 +17,7 @@ package com.squareup.sample.mainactivity
 
 import android.os.Bundle
 import androidx.appcompat.app.AppCompatActivity
+import androidx.lifecycle.lifecycleScope
 import androidx.test.espresso.IdlingResource
 import com.squareup.sample.authworkflow.AuthViewFactories
 import com.squareup.sample.container.SampleContainers
@@ -29,12 +30,11 @@ import com.squareup.workflow.ui.backstack.BackStackContainer
 import com.squareup.workflow.ui.modal.AlertContainer
 import com.squareup.workflow.ui.plus
 import com.squareup.workflow.ui.setContentWorkflow
-import io.reactivex.disposables.Disposables
+import kotlinx.coroutines.flow.collect
+import kotlinx.coroutines.launch
 import timber.log.Timber
 
 class MainActivity : AppCompatActivity() {
-  private var loggingSub = Disposables.disposed()
-
   private lateinit var component: MainComponent
 
   /** Exposed for use by espresso tests. */
@@ -67,16 +67,15 @@ class MainActivity : AppCompatActivity() {
         onResult = { finish() }
     )
 
-    loggingSub = workflowRunner.renderings.subscribe { Timber.d("rendering: %s", it) }
+    lifecycleScope.launch {
+      workflowRunner.renderings.collect {
+        Timber.d("rendering: %s", it)
+      }
+    }
   }
 
   override fun onRetainCustomNonConfigurationInstance(): Any = component
 
-  override fun onDestroy() {
-    loggingSub.dispose()
-    super.onDestroy()
-  }
-
   private companion object {
     val viewRegistry = SampleContainers +
         AuthViewFactories +

kotlin/workflow-ui/core-android/api/core-android.api

@@ -131,14 +131,14 @@ public abstract class com/squareup/workflow/ui/WorkflowFragment : androidx/fragm
 public final class com/squareup/workflow/ui/WorkflowLayout : android/widget/FrameLayout {
 	public fun <init> (Landroid/content/Context;Landroid/util/AttributeSet;)V
 	public synthetic fun <init> (Landroid/content/Context;Landroid/util/AttributeSet;ILkotlin/jvm/internal/DefaultConstructorMarker;)V
-	public final fun start (Lio/reactivex/Observable;Lcom/squareup/workflow/ui/ViewEnvironment;)V
-	public final fun start (Lio/reactivex/Observable;Lcom/squareup/workflow/ui/ViewRegistry;)V
+	public final fun start (Lkotlinx/coroutines/flow/Flow;Lcom/squareup/workflow/ui/ViewEnvironment;)V
+	public final fun start (Lkotlinx/coroutines/flow/Flow;Lcom/squareup/workflow/ui/ViewRegistry;)V
 }
 
 public abstract interface class com/squareup/workflow/ui/WorkflowRunner {
 	public static final field Companion Lcom/squareup/workflow/ui/WorkflowRunner$Companion;
-	public abstract fun getRenderings ()Lio/reactivex/Observable;
-	public abstract fun getResult ()Lio/reactivex/Maybe;
+	public abstract fun awaitResult (Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+	public abstract fun getRenderings ()Lkotlinx/coroutines/flow/Flow;
 }
 
 public final class com/squareup/workflow/ui/WorkflowRunner$Companion {

kotlin/workflow-ui/core-android/build.gradle.kts

@@ -38,15 +38,13 @@ dependencies {
 
   api(Dependencies.AndroidX.transition)
   api(Dependencies.Kotlin.Stdlib.jdk6)
-  api(Dependencies.RxJava2.rxjava2)
 
   implementation(Dependencies.AndroidX.activity)
   implementation(Dependencies.AndroidX.fragment)
-  implementation(Dependencies.AndroidX.Lifecycle.reactivestreams)
+  implementation(Dependencies.AndroidX.Lifecycle.ktx)
   implementation(Dependencies.AndroidX.savedstate)
   implementation(Dependencies.Kotlin.Coroutines.android)
   implementation(Dependencies.Kotlin.Coroutines.core)
-  implementation(Dependencies.Kotlin.Coroutines.rx2)
 
   testImplementation(Dependencies.Test.junit)
   testImplementation(Dependencies.Test.truth)

kotlin/workflow-ui/core-android/src/main/java/com/squareup/workflow/ui/WorkflowFragment.kt

@@ -21,11 +21,10 @@ import android.view.ViewGroup
 import androidx.fragment.app.Fragment
 import com.squareup.workflow.Workflow
 import com.squareup.workflow.ui.WorkflowRunner.Config
-import io.reactivex.Flowable
 
 /**
  * A [Fragment] that can run a workflow. Subclasses implement [onCreateWorkflow]
- * to configure themselves with a [Workflow], [ViewRegistry] and [inputs][Flowable].
+ * to configure themselves with a [Workflow], [ViewRegistry] and [inputs][Config.props].
  *
  * For a workflow with no inputs, or a static configuration, that's as simple as:
  *

kotlin/workflow-ui/core-android/src/main/java/com/squareup/workflow/ui/WorkflowLayout.kt

@@ -25,8 +25,13 @@ import android.view.View
 import android.view.ViewGroup
 import android.view.ViewGroup.LayoutParams.MATCH_PARENT
 import android.widget.FrameLayout
-import io.reactivex.Observable
-import io.reactivex.disposables.SerialDisposable
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.Job
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.launchIn
+import kotlinx.coroutines.flow.onEach
 
 /**
  * A view that can be driven by a [WorkflowRunner]. In most cases you'll use
@@ -49,7 +54,7 @@ class WorkflowLayout(
    * making that view the only child of this one.
    */
   fun start(
-    renderings: Observable<out Any>,
+    renderings: Flow<Any>,
     registry: ViewRegistry
   ) {
     start(renderings, ViewEnvironment(registry))
@@ -61,7 +66,7 @@ class WorkflowLayout(
    * making that view the only child of this one.
    */
   fun start(
-    renderings: Observable<out Any>,
+    renderings: Flow<Any>,
     environment: ViewEnvironment
   ) {
     val hintsWithDefaults = environment.withDefaultViewFactories()
@@ -133,18 +138,22 @@ class WorkflowLayout(
    * Subscribes [update] to [source] only while this [View] is attached to a window.
    */
   private fun <S : Any> View.takeWhileAttached(
-    source: Observable<S>,
+    source: Flow<S>,
     update: (S) -> Unit
   ) {
     val listener = object : OnAttachStateChangeListener {
-      var sub = SerialDisposable()
+      val scope = CoroutineScope(Dispatchers.Main.immediate)
+      var job: Job? = null
 
+      @OptIn(ExperimentalCoroutinesApi::class)
       override fun onViewAttachedToWindow(v: View?) {
-        sub.replace(source.subscribe { screen -> update(screen) })
+        job = source.onEach { screen -> update(screen) }
+            .launchIn(scope)
       }
 
       override fun onViewDetachedFromWindow(v: View?) {
-        sub.dispose()
+        job?.cancel()
+        job = null
       }
     }
 

kotlin/workflow-ui/core-android/src/main/java/com/squareup/workflow/ui/WorkflowRunner.kt

@@ -19,14 +19,11 @@ package com.squareup.workflow.ui
 
 import androidx.fragment.app.Fragment
 import androidx.fragment.app.FragmentActivity
-import androidx.lifecycle.Observer
 import androidx.lifecycle.ViewModelProvider
-import androidx.lifecycle.toLiveData
+import androidx.lifecycle.lifecycleScope
 import com.squareup.workflow.Workflow
 import com.squareup.workflow.diagnostic.WorkflowDiagnosticListener
 import com.squareup.workflow.ui.WorkflowRunner.Config
-import io.reactivex.Maybe
-import io.reactivex.Observable
 import kotlinx.coroutines.CoroutineDispatcher
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.flow.Flow
@@ -39,21 +36,22 @@ import kotlinx.coroutines.flow.flowOf
  * or subclass [WorkflowFragment] rather than instantiate a [WorkflowRunner] directly.
  */
 interface WorkflowRunner<out OutputT : Any> {
+
+  /**
+   * A stream of the rendering values emitted by the running [Workflow].
+   */
+  val renderings: Flow<Any>
+
   /**
-   * Provides the first (and only) [OutputT] value emitted by the workflow, or
-   * nothing if it is canceled before emitting.
+   * Returns the first (and only) [OutputT] value emitted by the workflow. Throws the cancellation
+   * exception if the workflow was cancelled before emitting.
    *
    * The output of the root workflow is treated as a result code, handy for use
    * as a sign that the host Activity or Fragment should be finished. Thus, once
-   * a value is emitted the workflow is ended its output value is reported through
+   * a value is emitted the workflow is ended and its output value is reported through
    * this field.
    */
-  val result: Maybe<out OutputT>
-
-  /**
-   * A stream of the rendering values emitted by the running [Workflow].
-   */
-  val renderings: Observable<out Any>
+  suspend fun awaitResult(): OutputT
 
   /**
    * @param diagnosticListener If non-null, will receive all diagnostic events from the workflow
@@ -161,9 +159,9 @@ fun <PropsT, OutputT : Any> FragmentActivity.setContentWorkflow(
     start(runner.renderings, viewEnvironment)
   }
 
-  runner.result.toFlowable()
-      .toLiveData()
-      .observe(this, Observer { result -> onResult(result) })
+  lifecycleScope.launchWhenStarted {
+    onResult(runner.awaitResult())
+  }
 
   this.setContentView(layout)
 

kotlin/workflow-ui/core-android/src/main/java/com/squareup/workflow/ui/WorkflowRunnerViewModel.kt

@@ -24,21 +24,23 @@ import com.squareup.workflow.Snapshot
 import com.squareup.workflow.WorkflowSession
 import com.squareup.workflow.launchWorkflowIn
 import com.squareup.workflow.ui.WorkflowRunner.Config
-import io.reactivex.Maybe
-import io.reactivex.Observable
+import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.ExperimentalCoroutinesApi
+import kotlinx.coroutines.async
 import kotlinx.coroutines.cancel
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.first
 import kotlinx.coroutines.flow.launchIn
 import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.flow.onEach
-import kotlinx.coroutines.rx2.asObservable
+import kotlinx.coroutines.launch
 import org.jetbrains.annotations.TestOnly
-import java.util.concurrent.CancellationException
 
 internal class WorkflowRunnerViewModel<OutputT : Any>(
   private val scope: CoroutineScope,
-  session: WorkflowSession<OutputT, Any>
+  private val session: WorkflowSession<OutputT, Any>
 ) : ViewModel(), WorkflowRunner<OutputT>, SavedStateProvider {
 
   internal class Factory<PropsT, OutputT : Any>(
@@ -64,28 +66,32 @@ internal class WorkflowRunnerViewModel<OutputT : Any>(
     }
   }
 
-  @Suppress("EXPERIMENTAL_API_USAGE")
-  override val result: Maybe<out OutputT> = session.outputs.asObservable()
-      .firstElement()
-      .doAfterTerminate {
-        scope.cancel(CancellationException("WorkflowRunnerViewModel delivered result"))
-      }
-      .cache()
+  private val result = scope.async {
+    session.outputs.first()
+  }
+
+  override suspend fun awaitResult(): OutputT = result.await()
 
   init {
     @Suppress("EXPERIMENTAL_API_USAGE")
     session.renderingsAndSnapshots
         .map { it.snapshot }
         .onEach { lastSnapshot = it }
         .launchIn(scope)
+
+    // Cancel the entire workflow runtime after the first output is emitted.
+    // Use the Unconfined dispatcher to ensure the cancellation happens as immediately as possible.
+    scope.launch(Dispatchers.Unconfined) {
+      result.join()
+      scope.cancel(CancellationException("WorkflowRunnerViewModel delivered result"))
+    }
   }
 
   private var lastSnapshot: Snapshot = Snapshot.EMPTY
 
   @OptIn(ExperimentalCoroutinesApi::class)
-  override val renderings: Observable<out Any> = session.renderingsAndSnapshots
+  override val renderings: Flow<Any> = session.renderingsAndSnapshots
       .map { it.rendering }
-      .asObservable()
 
   override fun onCleared() {
     scope.cancel(CancellationException("WorkflowRunnerViewModel cleared."))