Upgrade kotlin and coroutines dependencies to latest versions.

Closes #188 because the deprecated `cancel` method that took `Throwable` has been removed.

The issue where workflow subscriptions were disposed multiple times also seems to have been fixed.

The only real thing this "broke" was how legacy workflow's `switchMapState` handles channel cancellation,
now we have to be a bit more explicit. This is why channels shouldn't be used as streams…

Author: zach-klippenstein

kotlin/build.gradle

@@ -35,8 +35,8 @@ buildscript {
       'hamcrest': '1.3',
       'intellijAnnotations': '13.0',
       'junit': '4.12',
-      'kotlin': '1.3.21',
-      'kotlinCoroutines': '1.1.1',
+      'kotlin': '1.3.30',
+      'kotlinCoroutines': '1.2.0',
       'ktlintPlugin': '5.1.0',
       'mavenPublishPlugin': '0.6.0',
       'mockito': '2.7.5',

kotlin/legacy/legacy-workflow-core/src/main/java/com/squareup/workflow/legacy/WorkflowOperators.kt

@@ -17,6 +17,7 @@
 
 package com.squareup.workflow.legacy
 
+import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Deferred
 import kotlinx.coroutines.Dispatchers.Unconfined
@@ -93,7 +94,20 @@ fun <S1 : Any, S2 : Any, E : Any, O : Any> Workflow<S1, E, O>.switchMapState(
         // leave the consumeEach loop but the produce coroutine will wait for this child coroutine
         // to complete (structured concurrency) before closing the downstream channel.
         transformerJob = launch {
-          transform(upstreamState).toChannel(downstreamChannel)
+          try {
+            transform(upstreamState).toChannel(downstreamChannel)
+          } catch (e: Throwable) {
+            // Actual errors should be propagated, cancellation should interrupt forwarding
+            // the current transformed channel but allow the next one to continue.
+            // However, there may be multiple nested CancellationExceptions wrapping the actual
+            // exception, so we have to go digging.
+            val causeChain = generateSequence(e) { it.cause }
+            causeChain.firstOrNull { it !is CancellationException }
+                ?.let { realException ->
+                  downstreamChannel.close(realException)
+                  throw realException
+                }
+          }
         }
       }
     }
@@ -114,10 +128,13 @@ fun <S : Any, E : Any, O1 : Any, O2 : Any> Workflow<S, E, O1>.mapResult(
 
   // Propagate cancellation upstream.
   transformedResult.invokeOnCompletion { cause ->
-    if (cause != null) {
-      // TODO https://github.com/square/workflow/issues/188 Stop using parameterized cancel.
-      @Suppress("DEPRECATION")
+    this@mapResult.cancel(
+        if (cause is CancellationException) cause else CancellationException(null, cause)
+    )
+    if (cause is CancellationException) {
       this@mapResult.cancel(cause)
+    } else if (cause != null) {
+      this@mapResult.cancel(CancellationException(null, cause))
     }
   }
 

kotlin/legacy/legacy-workflow-core/src/test/java/com/squareup/workflow/legacy/CoroutineWorkflowTest.kt

@@ -21,6 +21,7 @@ import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineName
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers.Unconfined
+import kotlinx.coroutines.cancel
 import kotlinx.coroutines.channels.ReceiveChannel
 import kotlinx.coroutines.channels.consume
 import kotlinx.coroutines.suspendCancellableCoroutine
@@ -173,7 +174,7 @@ class CoroutineWorkflowTest : CoroutineScope {
     workflow.sendEvent(Unit)
   }
 
-  @Test fun `block gets original cancellation reason`() {
+  @Test fun `block gets original cancellation reason - null cause`() {
     lateinit var cancelReason: Throwable
     val workflow = workflow<Nothing, Nothing, Nothing> { _, _ ->
       suspendCancellableCoroutine<Nothing> { continuation ->
@@ -190,7 +191,7 @@ class CoroutineWorkflowTest : CoroutineScope {
   }
 
   @Suppress("DEPRECATION")
-  @Test fun `block gets original cancellation reason - deprecated cancel`() {
+  @Test fun `block gets original cancellation reason - non-null cause`() {
     lateinit var cancelReason: Throwable
     val workflow = workflow<Nothing, Nothing, Nothing> { _, _ ->
       suspendCancellableCoroutine<Nothing> { continuation ->
@@ -203,10 +204,16 @@ class CoroutineWorkflowTest : CoroutineScope {
     workflow.cancel(ExpectedException)
 
     assertTrue(cancelReason is CancellationException)
-    assertEquals(ExpectedException, cancelReason.cause)
+    // Search up the cause chain for the expected exception, since multiple CancellationExceptions
+    // may be chained together first.
+    val causeChain = generateSequence<Throwable>(cancelReason) { it.cause }
+    assertEquals(
+        1, causeChain.count { it === ExpectedException },
+        "Expected cancellation exception cause chain to include ExpectedException."
+    )
   }
 
   private companion object {
-    object ExpectedException : RuntimeException()
+    object ExpectedException : CancellationException()
   }
 }

kotlin/legacy/legacy-workflow-core/src/test/java/com/squareup/workflow/legacy/WorkflowOperatorsTest.kt

@@ -24,10 +24,12 @@ import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.channels.consume
 import kotlinx.coroutines.channels.produce
 import kotlinx.coroutines.isActive
+import kotlinx.coroutines.runBlocking
 import kotlinx.coroutines.suspendCancellableCoroutine
 import kotlin.test.BeforeTest
 import kotlin.test.Test
 import kotlin.test.assertEquals
+import kotlin.test.assertFails
 import kotlin.test.assertFailsWith
 import kotlin.test.assertFalse
 import kotlin.test.assertNull
@@ -81,9 +83,7 @@ class WorkflowOperatorsTest {
       it.toString()
     }
 
-    // TODO https://github.com/square/workflow/issues/188 Stop using parameterized cancel.
-    @Suppress("DEPRECATION")
-    withAdaptedEvents.cancel(ExpectedException())
+    withAdaptedEvents.cancel(CancellationException(null, ExpectedException()))
 
     assertTrue(sourceCancellation is CancellationException)
   }
@@ -161,9 +161,7 @@ class WorkflowOperatorsTest {
     val withMappedStates = source.mapState { it }
 
     assertNull(sourceCancellation)
-    // TODO https://github.com/square/workflow/issues/188 Stop using parameterized cancel.
-    @Suppress("DEPRECATION")
-    withMappedStates.cancel(ExpectedException())
+    withMappedStates.cancel(CancellationException(null, ExpectedException()))
     assertTrue(sourceCancellation is CancellationException)
   }
 
@@ -308,7 +306,16 @@ class WorkflowOperatorsTest {
         .consume {
           assertFalse(withMappedStates.isCompleted)
           withMappedStates.sendEvent(Unit)
-          assertFailsWith<ExpectedException> { poll() }
+          assertFails { runBlocking { receive() } }
+              .also { error ->
+                // Search up the cause chain for the expected exception, since multiple CancellationExceptions
+                // may be chained together first.
+                val causeChain = generateSequence(error) { it.cause }
+                assertEquals(
+                    1, causeChain.count { it is ExpectedException },
+                    "Expected cancellation exception cause chain to include ExpectedException."
+                )
+              }
           // Exception is not sent through the result.
           assertEquals(Unit, withMappedStates.getCompleted())
         }
@@ -321,17 +328,24 @@ class WorkflowOperatorsTest {
     }
     val withMappedStates: Workflow<Int, Unit, Unit> = source.switchMapState {
       Channel<Int>().apply {
-        // TODO https://github.com/square/workflow/issues/188 Stop using parameterized cancel.
-        @Suppress("DEPRECATION")
-        cancel(ExpectedException())
+        cancel(CancellationException(null, ExpectedException()))
       }
     }
 
     withMappedStates.openSubscriptionToState()
         .consume {
           assertFalse(withMappedStates.isCompleted)
           withMappedStates.sendEvent(Unit)
-          assertFailsWith<ExpectedException> { poll() }
+          assertFails { runBlocking { receive() } }
+              .also { error ->
+                // Search up the cause chain for the expected exception, since multiple CancellationExceptions
+                // may be chained together first.
+                val causeChain = generateSequence(error) { it.cause }
+                assertEquals(
+                    1, causeChain.count { it is ExpectedException },
+                    "Expected cancellation exception cause chain to include ExpectedException."
+                )
+              }
           // Exception is not sent through the result.
           assertEquals(Unit, withMappedStates.getCompleted())
         }
@@ -371,9 +385,7 @@ class WorkflowOperatorsTest {
           withMappedStates.sendEvent(Unit)
 
           assertFalse(transformedChannel.isClosedForSend)
-          // TODO https://github.com/square/workflow/issues/188 Stop using parameterized cancel.
-          @Suppress("DEPRECATION")
-          this.cancel(ExpectedException())
+          this.cancel(CancellationException(null, ExpectedException()))
           assertTrue(transformedChannel.isClosedForSend)
         }
   }
@@ -390,9 +402,7 @@ class WorkflowOperatorsTest {
     val withMappedStates = source.switchMapState { produce { send(it) } }
 
     assertNull(sourceCancellation)
-    // TODO https://github.com/square/workflow/issues/188 Stop using parameterized cancel.
-    @Suppress("DEPRECATION")
-    withMappedStates.cancel(ExpectedException())
+    withMappedStates.cancel(CancellationException(null, ExpectedException()))
     assertTrue(sourceCancellation is CancellationException)
   }
 
@@ -450,9 +460,7 @@ class WorkflowOperatorsTest {
     val withMappedResult = source.mapResult { it }
 
     assertNull(sourceCancellation)
-    // TODO https://github.com/square/workflow/issues/188 Stop using parameterized cancel.
-    @Suppress("DEPRECATION")
-    withMappedResult.cancel(ExpectedException())
+    withMappedResult.cancel(CancellationException(null, ExpectedException()))
     assertTrue(sourceCancellation is CancellationException)
   }
 }

kotlin/legacy/legacy-workflow-rx2/src/main/java/com/squareup/workflow/legacy/rx2/EventChannel.kt

@@ -123,7 +123,7 @@ fun <E : Any> ReceiveChannel<E>.asEventChannel() = object : EventChannel<E> {
             .also { selectionJob.cancel() }
       } catch (cancellation: CancellationException) {
         val cause = cancellation.cause
-        if (cause == null) {
+        if (cause == null || cause is CancellationException) {
           // The select was cancelled normally, which means the workflow was abandoned and we're
           // about to get unsubscribed from. Don't propagate the error, just never emit/return.
           suspendCoroutine<Nothing> { }

kotlin/legacy/legacy-workflow-rx2/src/test/java/com/squareup/workflow/legacy/rx2/WorkflowOperatorsTest.kt

@@ -78,8 +78,6 @@ class WorkflowOperatorsTest {
     states.cancel()
 
     assertThat(subscribeCount).isEqualTo(1)
-    // For some reason the observable is actually disposed twice. Seems like a coroutines bug, but
-    // Disposable.dispose() is an idempotent operation so it should be fine.
-    assertThat(disposeCount).isEqualTo(2)
+    assertThat(disposeCount).isEqualTo(1)
   }
 }

kotlin/samples/tictactoe/android/build.gradle

@@ -27,6 +27,11 @@ android {
     versionCode 1
     versionName "1.0.0"
   }
+
+  // See https://github.com/Kotlin/kotlinx.coroutines/issues/1064#issuecomment-479412940
+  packagingOptions {
+    exclude 'META-INF/atomicfu.kotlin_module'
+  }
 }
 
 dependencies {

kotlin/workflow-runtime/src/main/java/com/squareup/workflow/WorkflowHost.kt

@@ -22,6 +22,7 @@ import com.squareup.workflow.WorkflowHost.Update
 import com.squareup.workflow.internal.WorkflowId
 import com.squareup.workflow.internal.WorkflowNode
 import com.squareup.workflow.internal.id
+import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.cancel
 import kotlinx.coroutines.channels.ReceiveChannel
 import kotlinx.coroutines.channels.produce
@@ -149,9 +150,7 @@ internal fun <I : Any, O : Any, R : Any> WorkflowNode<I, *, O, R>.start(
   } catch (e: Throwable) {
     // For some reason the exception gets masked if we don't explicitly pass it to cancel the
     // producer coroutine ourselves here.
-    // TODO https://github.com/square/workflow/issues/188 Stop using parameterized cancel.
-    @Suppress("DEPRECATION")
-    coroutineContext.cancel(e)
+    coroutineContext.cancel(if (e is CancellationException) e else CancellationException(null, e))
     throw e
   } finally {
     // There's a potential race condition if the producer coroutine is cancelled before it has a chance

kotlin/workflow-runtime/src/test/java/com/squareup/workflow/internal/ChannelUpdatesTest.kt

@@ -20,6 +20,7 @@ package com.squareup.workflow.internal
 import com.squareup.workflow.util.ChannelUpdate
 import com.squareup.workflow.util.ChannelUpdate.Closed
 import com.squareup.workflow.util.ChannelUpdate.Value
+import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.GlobalScope
 import kotlinx.coroutines.async
@@ -82,16 +83,22 @@ class ChannelUpdatesTest {
 
   @Test fun `handles error`() {
     val channel = Channel<String>()
-    // TODO https://github.com/square/workflow/issues/188 Stop using parameterized cancel.
-    @Suppress("DEPRECATION")
-    channel.cancel(ExpectedException())
+    channel.cancel(CancellationException(null, ExpectedException()))
 
-    assertFailsWith<ExpectedException> {
+    assertFailsWith<CancellationException> {
       runBlocking {
         select<ChannelUpdate<String>> {
           onChannelUpdate(channel) { it }
         }
       }
+    }.also { error ->
+      // Search up the cause chain for the expected exception, since multiple CancellationExceptions
+      // may be chained together first.
+      val causeChain = generateSequence<Throwable>(error) { it.cause }
+      assertEquals(
+          1, causeChain.count { it is ExpectedException },
+          "Expected cancellation exception cause chain to include original cause."
+      )
     }
   }
 

kotlin/workflow-runtime/src/test/java/com/squareup/workflow/internal/RealWorkflowContextTest.kt

@@ -23,7 +23,7 @@ import com.squareup.workflow.Workflow
 import com.squareup.workflow.WorkflowAction.Companion.emitOutput
 import com.squareup.workflow.WorkflowAction.Companion.noop
 import com.squareup.workflow.WorkflowContext
-import com.squareup.workflow.compose
+import com.squareup.workflow.composeChild
 import com.squareup.workflow.internal.Behavior.WorkflowOutputCase
 import com.squareup.workflow.internal.RealWorkflowContext.Composer
 import com.squareup.workflow.internal.RealWorkflowContextTest.TestComposer.Rendering
@@ -149,7 +149,7 @@ class RealWorkflowContextTest {
       context.onReceive<Unit>({ fail() }, Unit::class.starProjectedType) { fail() }
     }
     val child = Workflow.stateless<Nothing, Unit> { fail() }
-    assertFailsWith<IllegalStateException> { context.compose(child) }
+    assertFailsWith<IllegalStateException> { context.composeChild(child) }
     assertFailsWith<IllegalStateException> { context.buildBehavior() }
   }
 }

kotlin/workflow-rx2/src/test/java/com/squareup/workflow/rx2/SubscriptionsTest.kt

@@ -101,9 +101,7 @@ class SubscriptionsTest {
       }
 
       assertEquals(1, workflow.subscriptions)
-      // For some reason the observable is actually disposed twice. Seems like a coroutines bug, but
-      // Disposable.dispose() is an idempotent operation so it should be fine.
-      assertEquals(2, workflow.disposals)
+      assertEquals(1, workflow.disposals)
     }
   }
 
@@ -116,9 +114,7 @@ class SubscriptionsTest {
       }
 
       assertEquals(1, workflow.subscriptions)
-      // For some reason the observable is actually disposed twice. Seems like a coroutines bug, but
-      // Disposable.dispose() is an idempotent operation so it should be fine.
-      assertEquals(2, workflow.disposals)
+      assertEquals(1, workflow.disposals)
     }
   }
 
@@ -148,15 +144,13 @@ class SubscriptionsTest {
 
       host.withNextRendering { setSubscribed ->
         assertEquals(1, workflow.subscriptions)
-        // For some reason the observable is actually disposed twice. Seems like a coroutines bug,
-        // but Disposable.dispose() is an idempotent operation so it should be fine.
-        assertEquals(2, workflow.disposals)
+        assertEquals(1, workflow.disposals)
         setSubscribed(true)
       }
 
       host.withNextRendering {
         assertEquals(2, workflow.subscriptions)
-        assertEquals(2, workflow.disposals)
+        assertEquals(1, workflow.disposals)
       }
     }
   }