Ensure posted events from the ViewModel are consumed (once) by the UI

Inspired from https://github.com/Kotlin/kotlinx.coroutines/issues/3002

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)

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)
                         }

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)

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)
 }

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) }
+}

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)
             }
         }

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)
                         }