7 files changed, 29 insertions, 14 deletions
diff --git a/ui/kotlinx-coroutines-javafx/src/JavaFxConvert.kt b/ui/kotlinx-coroutines-javafx/src/JavaFxConvert.kt
index 1cbf9b6f..ebeaa3b8 100644
--- a/ui/kotlinx-coroutines-javafx/src/JavaFxConvert.kt
+++ b/ui/kotlinx-coroutines-javafx/src/JavaFxConvert.kt
@@ -4,10 +4,9 @@
 
 package kotlinx.coroutines.javafx
 
-import javafx.beans.value.ChangeListener
-import javafx.beans.value.ObservableValue
+import javafx.beans.value.*
 import kotlinx.coroutines.*
-import kotlinx.coroutines.channels.awaitClose
+import kotlinx.coroutines.channels.*
 import kotlinx.coroutines.flow.*
 
 /**
@@ -27,11 +26,11 @@ import kotlinx.coroutines.flow.*
 @ExperimentalCoroutinesApi // Since 1.3.x
 public fun <T> ObservableValue<T>.asFlow(): Flow<T> = callbackFlow<T> {
     val listener = ChangeListener<T> { _, _, newValue ->
-        try {
-            offer(newValue)
-        } catch (e: CancellationException) {
-            // In case the event fires after the channel is closed
-        }
+        /*
+         * Do not propagate the exception to the ObservableValue, it
+         * already should've been handled by the downstream
+         */
+        trySend(newValue)
     }
     addListener(listener)
     send(value)
diff --git a/ui/kotlinx-coroutines-javafx/test/JavaFxObservableAsFlowTest.kt b/ui/kotlinx-coroutines-javafx/test/JavaFxObservableAsFlowTest.kt
index 69640501..bc40b0fd 100644
--- a/ui/kotlinx-coroutines-javafx/test/JavaFxObservableAsFlowTest.kt
+++ b/ui/kotlinx-coroutines-javafx/test/JavaFxObservableAsFlowTest.kt
@@ -83,4 +83,20 @@ class JavaFxObservableAsFlowTest : TestBase() {
         }
     }
 
+    @Test
+    fun testIntermediateCrash() = runTest {
+        if (!initPlatform()) {
+            println("Skipping JavaFxTest in headless environment")
+            return@runTest // ignore test in headless environments
+        }
+
+        val property = SimpleIntegerProperty(0)
+
+        assertFailsWith<TestException> {
+            property.asFlow().onEach {
+                yield()
+                throw TestException()
+            }.collect()
+        }
+    }
 }
diff --git a/ui/kotlinx-coroutines-javafx/test/guide/example-ui-actor-02.kt b/ui/kotlinx-coroutines-javafx/test/guide/example-ui-actor-02.kt
index 51e94779..ec8a09f9 100644
--- a/ui/kotlinx-coroutines-javafx/test/guide/example-ui-actor-02.kt
+++ b/ui/kotlinx-coroutines-javafx/test/guide/example-ui-actor-02.kt
@@ -67,6 +67,6 @@ fun Node.onClick(action: suspend (MouseEvent) -> Unit) {
     }
     // install a listener to offer events to this actor
     onMouseClicked = EventHandler { event ->
-        eventActor.offer(event)
+        eventActor.trySend(event)
     }
 }
diff --git a/ui/kotlinx-coroutines-javafx/test/guide/example-ui-actor-03.kt b/ui/kotlinx-coroutines-javafx/test/guide/example-ui-actor-03.kt
index 81371678..aa152b79 100644
--- a/ui/kotlinx-coroutines-javafx/test/guide/example-ui-actor-03.kt
+++ b/ui/kotlinx-coroutines-javafx/test/guide/example-ui-actor-03.kt
@@ -65,8 +65,8 @@ fun Node.onClick(action: suspend (MouseEvent) -> Unit) {
     val eventActor = GlobalScope.actor<MouseEvent>(Dispatchers.Main, capacity = Channel.CONFLATED) { // <--- Changed here
         for (event in channel) action(event) // pass event to action
     }
-    // install a listener to offer events to this actor
+    // install a listener to send events to this actor
     onMouseClicked = EventHandler { event ->
-        eventActor.offer(event)
+        eventActor.trySend(event)
     }
 }
diff --git a/ui/kotlinx-coroutines-javafx/test/guide/example-ui-blocking-01.kt b/ui/kotlinx-coroutines-javafx/test/guide/example-ui-blocking-01.kt
index ea5ac90a..0c89ea70 100644
--- a/ui/kotlinx-coroutines-javafx/test/guide/example-ui-blocking-01.kt
+++ b/ui/kotlinx-coroutines-javafx/test/guide/example-ui-blocking-01.kt
@@ -55,7 +55,7 @@ fun Node.onClick(action: suspend (MouseEvent) -> Unit) {
         for (event in channel) action(event) // pass event to action
     }
     onMouseClicked = EventHandler { event ->
-        eventActor.offer(event)
+        eventActor.trySend(event)
     }
 }
 
diff --git a/ui/kotlinx-coroutines-javafx/test/guide/example-ui-blocking-02.kt b/ui/kotlinx-coroutines-javafx/test/guide/example-ui-blocking-02.kt
index 504f2ee6..6e8b984a 100644
--- a/ui/kotlinx-coroutines-javafx/test/guide/example-ui-blocking-02.kt
+++ b/ui/kotlinx-coroutines-javafx/test/guide/example-ui-blocking-02.kt
@@ -55,7 +55,7 @@ fun Node.onClick(action: suspend (MouseEvent) -> Unit) {
         for (event in channel) action(event) // pass event to action
     }
     onMouseClicked = EventHandler { event ->
-        eventActor.offer(event)
+        eventActor.trySend(event)
     }
 }
 
diff --git a/ui/kotlinx-coroutines-javafx/test/guide/example-ui-blocking-03.kt b/ui/kotlinx-coroutines-javafx/test/guide/example-ui-blocking-03.kt
index 0e115367..3ff5d7d5 100644
--- a/ui/kotlinx-coroutines-javafx/test/guide/example-ui-blocking-03.kt
+++ b/ui/kotlinx-coroutines-javafx/test/guide/example-ui-blocking-03.kt
@@ -55,7 +55,7 @@ fun Node.onClick(action: suspend (MouseEvent) -> Unit) {
         for (event in channel) action(event) // pass event to action
     }
     onMouseClicked = EventHandler { event ->
-        eventActor.offer(event)
+        eventActor.trySend(event)
     }
 }