Issue bumble-tech#296 Dispose NodeConnector intake in onDestroy

LachlanMcKee · Dec 4, 2022 · d4da1b5 · d4da1b5
1 parent e636b6b
commit d4da1b5
Show file tree

Hide file tree

Showing 2 changed files with 5 additions and 3 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -2,7 +2,7 @@
 
 ## Pending changes
 
-–
+- [#296](https://github.com/bumble-tech/appyx/issues/296) – **Fixed**: `NodeConnector.intake` is now disposed when a `Node` is destroyed.
 
 ---
 

diff --git a/...ies/interop-rx2/src/main/kotlin/com/bumble/appyx/interop/rx2/connectable/NodeConnector.kt b/...ies/interop-rx2/src/main/kotlin/com/bumble/appyx/interop/rx2/connectable/NodeConnector.kt
@@ -5,6 +5,7 @@ import com.bumble.appyx.core.lifecycle.subscribe
 import com.jakewharton.rxrelay2.PublishRelay
 import com.jakewharton.rxrelay2.Relay
 import io.reactivex.Observer
+import io.reactivex.disposables.Disposable
 
 class NodeConnector<Input, Output>(
     override val input: Relay<Input> = PublishRelay.create(),
@@ -15,6 +16,7 @@ class NodeConnector<Input, Output>(
     private val exhaust: Relay<Output> = PublishRelay.create()
     private var isFlushed = false
     private val outputCache = mutableListOf<Output>()
+    private var intakeDisposable: Disposable? = null
 
     override val output: Relay<Output> = object : Relay<Output>() {
 
@@ -31,7 +33,7 @@ class NodeConnector<Input, Output>(
     }
 
     override fun onCreate(lifecycle: Lifecycle) {
-        lifecycle.subscribe(onCreate = { flushOutputCache() })
+        lifecycle.subscribe(onCreate = { flushOutputCache() }, onDestroy = { intakeDisposable?.dispose() })
     }
 
     private val cacheSubscription = intake.subscribe {
@@ -55,7 +57,7 @@ class NodeConnector<Input, Output>(
     }
 
     private fun switchToExhaust() {
-        intake.subscribe { exhaust.accept(it) }
+        intakeDisposable = intake.subscribe { exhaust.accept(it) }
         cacheSubscription.dispose()
     }
 }