Rx.createEmitFlow() now creates a flow with unlimited buffer, same …

…as RxJava's `PublishSubject`
diffplug · Jan 31, 2025 · 6bac4d1 · 6bac4d1
1 parent 4b41ebf
commit 6bac4d1
Show file tree

Hide file tree

Showing 2 changed files with 5 additions and 1 deletion.
diff --git a/CHANGES.md b/CHANGES.md
@@ -1,6 +1,8 @@
 # DurianRx releases
 
 ## [Unreleased]
+### Fixed
+- `Rx.createEmitFlow()` now creates a flow with unlimited buffer, same as RxJava's `PublishSubject`.
 
 ## [5.0.1] - 2025-01-26
 ### Changed

diff --git a/src/main/java/com/diffplug/common/rx/Rx.kt b/src/main/java/com/diffplug/common/rx/Rx.kt
@@ -36,6 +36,7 @@ import kotlinx.coroutines.Deferred
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.asCoroutineDispatcher
 import kotlinx.coroutines.channels.BufferOverflow
+import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.merge
@@ -102,7 +103,8 @@ import kotlinx.coroutines.flow.merge
 object Rx {
 	@JvmStatic
 	fun <T> createEmitFlow() =
-			MutableSharedFlow<T>(replay = 0, extraBufferCapacity = 1, BufferOverflow.SUSPEND)
+			MutableSharedFlow<T>(
+					replay = 0, extraBufferCapacity = Channel.UNLIMITED, BufferOverflow.SUSPEND)
 
 	@JvmStatic
 	fun <T> emit(flow: MutableSharedFlow<T>, value: T) {