Change after PR review

changelog.d/6278.bugfix

@@ -0,0 +1 @@
+Fix regression on EventInsertLiveObserver getting blocked so there is no event being processed anymore.

matrix-sdk-android/src/main/java/org/matrix/android/sdk/internal/database/EventInsertLiveObserver.kt

@@ -19,10 +19,9 @@ package org.matrix.android.sdk.internal.database
 import com.zhuinden.monarchy.Monarchy
 import io.realm.RealmConfiguration
 import io.realm.RealmResults
-import kotlinx.coroutines.flow.MutableSharedFlow
-import kotlinx.coroutines.flow.launchIn
-import kotlinx.coroutines.flow.onEach
 import kotlinx.coroutines.launch
+import kotlinx.coroutines.sync.Mutex
+import kotlinx.coroutines.sync.withLock
 import org.matrix.android.sdk.internal.database.mapper.asDomain
 import org.matrix.android.sdk.internal.database.model.EventEntity
 import org.matrix.android.sdk.internal.database.model.EventInsertEntity
@@ -39,26 +38,18 @@ internal class EventInsertLiveObserver @Inject constructor(
 ) :
         RealmLiveEntityObserver<EventInsertEntity>(realmConfiguration) {
 
+    private val lock = Mutex()
+
     override val query = Monarchy.Query {
         it.where(EventInsertEntity::class.java).equalTo(EventInsertEntityFields.CAN_BE_PROCESSED, true)
     }
 
-    private val onResultsChangedFlow = MutableSharedFlow<RealmResults<EventInsertEntity>>()
-
-    init {
-        onResultsChangedFlow
-                .onEach { handleChange(it) }
-                .launchIn(observerScope)
-    }
-
     override fun onChange(results: RealmResults<EventInsertEntity>) {
+        observerScope.launch {
+            lock.withLock {
                 if (!results.isLoaded || results.isEmpty()) {
-            return
+                    return@withLock
                 }
-        observerScope.launch { onResultsChangedFlow.emit(results) }
-    }
-
-    private suspend fun handleChange(results: RealmResults<EventInsertEntity>) {
                 val idsToDeleteAfterProcess = ArrayList<String>()
                 val filteredEvents = ArrayList<EventInsertEntity>(results.size)
                 Timber.v("EventInsertEntity updated with ${results.size} results in db")
@@ -75,7 +66,6 @@ internal class EventInsertLiveObserver @Inject constructor(
                     }
                     idsToDeleteAfterProcess.add(it.eventId)
                 }
-
                 awaitTransaction(realmConfiguration) { realm ->
                     Timber.v("##Transaction: There are ${filteredEvents.size} events to process ")
                     filteredEvents.forEach { eventInsert ->
@@ -99,6 +89,8 @@ internal class EventInsertLiveObserver @Inject constructor(
                 }
                 processors.forEach { it.onPostProcess() }
             }
+        }
+    }
 
     private fun shouldProcess(eventInsertEntity: EventInsertEntity): Boolean {
         return processors.any {