diff --git a/vector/src/main/java/im/vector/app/core/platform/VectorBaseActivity.kt b/vector/src/main/java/im/vector/app/core/platform/VectorBaseActivity.kt
index a87eb92b13..1e29dfff5e 100644
--- a/vector/src/main/java/im/vector/app/core/platform/VectorBaseActivity.kt
+++ b/vector/src/main/java/im/vector/app/core/platform/VectorBaseActivity.kt
@@ -128,9 +128,11 @@ abstract class VectorBaseActivity<VB : ViewBinding> : AppCompatActivity(), Maver
fun <T : VectorViewEvents> VectorViewModel<*, *, T>.observeViewEvents(
observer: (T) -> Unit,
) {
+ val tag = this@VectorBaseActivity::class.simpleName.toString()
lifecycleScope.launch {
repeatOnLifecycle(Lifecycle.State.RESUMED) {
- viewEvents.stream()
+ viewEvents
+ .stream(tag)
.collect {
hideWaitingView()
observer(it)
diff --git a/vector/src/main/java/im/vector/app/core/platform/VectorBaseBottomSheetDialogFragment.kt b/vector/src/main/java/im/vector/app/core/platform/VectorBaseBottomSheetDialogFragment.kt
index ad7a86c899..a44fb1c9ac 100644
--- a/vector/src/main/java/im/vector/app/core/platform/VectorBaseBottomSheetDialogFragment.kt
+++ b/vector/src/main/java/im/vector/app/core/platform/VectorBaseBottomSheetDialogFragment.kt
@@ -205,9 +205,11 @@ abstract class VectorBaseBottomSheetDialogFragment<VB : ViewBinding> : BottomShe
protected fun <T : VectorViewEvents> VectorViewModel<*, *, T>.observeViewEvents(
observer: (T) -> Unit,
) {
+ val tag = this@VectorBaseBottomSheetDialogFragment::class.simpleName.toString()
lifecycleScope.launch {
repeatOnLifecycle(Lifecycle.State.RESUMED) {
- viewEvents.stream()
+ viewEvents
+ .stream(tag)
.collect {
observer(it)
}
diff --git a/vector/src/main/java/im/vector/app/core/platform/VectorBaseFragment.kt b/vector/src/main/java/im/vector/app/core/platform/VectorBaseFragment.kt
index 9f79db9c66..a82cef54e5 100644
--- a/vector/src/main/java/im/vector/app/core/platform/VectorBaseFragment.kt
+++ b/vector/src/main/java/im/vector/app/core/platform/VectorBaseFragment.kt
@@ -277,9 +277,11 @@ abstract class VectorBaseFragment<VB : ViewBinding> : Fragment(), MavericksView
protected fun <T : VectorViewEvents> VectorViewModel<*, *, T>.observeViewEvents(
observer: (T) -> Unit,
) {
+ val tag = this@VectorBaseFragment::class.simpleName.toString()
lifecycleScope.launch {
repeatOnLifecycle(Lifecycle.State.RESUMED) {
- viewEvents.stream()
+ viewEvents
+ .stream(tag)
.collect {
dismissLoadingDialog()
observer(it)
diff --git a/vector/src/main/java/im/vector/app/core/platform/VectorViewModel.kt b/vector/src/main/java/im/vector/app/core/platform/VectorViewModel.kt
index c9d58f9545..3dd38c455f 100644
--- a/vector/src/main/java/im/vector/app/core/platform/VectorViewModel.kt
+++ b/vector/src/main/java/im/vector/app/core/platform/VectorViewModel.kt
@@ -18,15 +18,16 @@ package im.vector.app.core.platform
import com.airbnb.mvrx.MavericksState
import com.airbnb.mvrx.MavericksViewModel
-import im.vector.app.core.utils.DataSource
-import im.vector.app.core.utils.PublishDataSource
+import im.vector.app.core.utils.EventQueue
+import im.vector.app.core.utils.SharedEvents
abstract class VectorViewModel<S : MavericksState, VA : VectorViewModelAction, VE : VectorViewEvents>(initialState: S) :
MavericksViewModel<S>(initialState) {
// Used to post transient events to the View
- protected val _viewEvents = PublishDataSource<VE>()
- val viewEvents: DataSource<VE> = _viewEvents
+ protected val _viewEvents = EventQueue<VE>(capacity = 64)
+ val viewEvents: SharedEvents<VE>
+ get() = _viewEvents
abstract fun handle(action: VA)
}
diff --git a/vector/src/main/java/im/vector/app/core/utils/SharedEvent.kt b/vector/src/main/java/im/vector/app/core/utils/SharedEvent.kt
new file mode 100644
index 0000000000..e712769c48
--- /dev/null
+++ b/vector/src/main/java/im/vector/app/core/utils/SharedEvent.kt
@@ -0,0 +1,58 @@
+/*
+ * Copyright 2022 New Vector Ltd
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package im.vector.app.core.utils
+
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.MutableSharedFlow
+import kotlinx.coroutines.flow.transform
+import java.util.concurrent.CopyOnWriteArraySet
+
+interface SharedEvents<out T> {
+ fun stream(consumerId: String): Flow<T>
+}
+
+class EventQueue<T>(capacity: Int) : SharedEvents<T> {
+
+ private val innerQueue = MutableSharedFlow<OneTimeEvent<T>>(replay = capacity)
+
+ fun post(event: T) {
+ innerQueue.tryEmit(OneTimeEvent(event))
+ }
+
+ override fun stream(consumerId: String): Flow<T> = innerQueue.filterNotHandledBy(consumerId)
+}
+
+/**
+ * Event designed to be delivered only once to a concrete entity,
+ * but it can also be delivered to multiple different entities.
+ *
+ * Keeps track of who has already handled its content.
+ */
+private class OneTimeEvent<out T>(private val content: T) {
+
+ private val handlers = CopyOnWriteArraySet<String>()
+
+ /**
+ * @param asker Used to identify, whether this "asker" has already handled this Event.
+ * @return Event content or null if it has been already handled by asker
+ */
+ fun getIfNotHandled(asker: String): T? = if (handlers.add(asker)) content else null
+}
+
+private fun <T> Flow<OneTimeEvent<T>>.filterNotHandledBy(consumerId: String): Flow<T> = transform { event ->
+ event.getIfNotHandled(consumerId)?.let { emit(it) }
+}
diff --git a/vector/src/main/java/im/vector/app/features/media/VectorAttachmentViewerActivity.kt b/vector/src/main/java/im/vector/app/features/media/VectorAttachmentViewerActivity.kt
index 9f9488e35d..0d240b376b 100644
--- a/vector/src/main/java/im/vector/app/features/media/VectorAttachmentViewerActivity.kt
+++ b/vector/src/main/java/im/vector/app/features/media/VectorAttachmentViewerActivity.kt
@@ -239,11 +239,12 @@ class VectorAttachmentViewerActivity : AttachmentViewerActivity(), AttachmentInt
}
private fun observeViewEvents() {
+ val tag = this::class.simpleName.toString()
lifecycleScope.launch {
repeatOnLifecycle(Lifecycle.State.RESUMED) {
viewModel
.viewEvents
- .stream()
+ .stream(tag)
.collect(::handleViewEvents)
}
}
diff --git a/vector/src/main/java/im/vector/app/features/settings/VectorSettingsBaseFragment.kt b/vector/src/main/java/im/vector/app/features/settings/VectorSettingsBaseFragment.kt
index 6299d8962d..724807a81e 100644
--- a/vector/src/main/java/im/vector/app/features/settings/VectorSettingsBaseFragment.kt
+++ b/vector/src/main/java/im/vector/app/features/settings/VectorSettingsBaseFragment.kt
@@ -72,10 +72,11 @@ abstract class VectorSettingsBaseFragment : PreferenceFragmentCompat(), Maverick
protected fun <T : VectorViewEvents> VectorViewModel<*, *, T>.observeViewEvents(
observer: (T) -> Unit,
) {
+ val tag = this@VectorSettingsBaseFragment::class.simpleName.toString()
lifecycleScope.launch {
- repeatOnLifecycle(state) {
repeatOnLifecycle(Lifecycle.State.RESUMED) {
- viewEvents.stream()
+ viewEvents
+ .stream(tag)
.collect {
observer(it)
}