Adds buffer to the message loading flow

MAILAND-2739

app/src/main/java/ch/protonmail/android/mailbox/presentation/MailboxViewModel.kt

@@ -33,6 +33,8 @@ import ch.protonmail.android.core.UserManager
 import ch.protonmail.android.data.ContactsRepository
 import ch.protonmail.android.data.local.model.Message
 import ch.protonmail.android.domain.LoadMoreFlow
+import ch.protonmail.android.domain.loadMoreBuffer
+import ch.protonmail.android.domain.loadMoreCombine
 import ch.protonmail.android.domain.loadMoreCombineTransform
 import ch.protonmail.android.domain.loadMoreMap
 import ch.protonmail.android.drawer.presentation.mapper.DrawerFoldersAndLabelsSectionUiModelMapper
@@ -73,6 +75,7 @@ import kotlinx.coroutines.flow.MutableSharedFlow
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.SharedFlow
 import kotlinx.coroutines.flow.asStateFlow
+import kotlinx.coroutines.flow.buffer
 import kotlinx.coroutines.flow.catch
 import kotlinx.coroutines.flow.combine
 import kotlinx.coroutines.flow.combineTransform
@@ -231,11 +234,7 @@ internal class MailboxViewModel @Inject constructor(
             .onEach {
                 mutableMailboxState.value = MailboxState.Loading
             }
-            .flatMapLatest { parameters ->
-                val location = parameters.first
-                val labelId = parameters.second
-                val userId = parameters.third
-
+            .flatMapLatest { (location, labelId, userId) ->
                 mailboxStateFlow = if (conversationModeEnabled(location)) {
                     Timber.v("Getting conversations for $location, label: $labelId, user: $userId")
                     conversationsAsMailboxItems(location, labelId, userId)
@@ -327,15 +326,14 @@ internal class MailboxViewModel @Inject constructor(
         Timber.v("conversationsAsMailboxItems locationId: $locationId")
         var isFirstData = true
         var hasReceivedFirstApiRefresh: Boolean? = null
-        return loadMoreCombineTransform<List<Label>, GetConversationsResult, Pair<List<Label>, GetConversationsResult>>(
+        return loadMoreCombine(
             observeLabels(userId),
             observeConversationsByLocation(
                 userId,
                 locationId
             )
-        ) { conversations, labels ->
-            emit(conversations to labels)
-        }
+        ) { labels, conversations -> labels to conversations }
+            .loadMoreBuffer()
             .loadMoreMap { pair ->
                 val labels = pair.first
                 when (val result = pair.second) {
@@ -380,49 +378,49 @@ internal class MailboxViewModel @Inject constructor(
         Timber.v("messagesAsMailboxItems location: $location, labelId: $labelId")
         var isFirstData = true
         var hasReceivedFirstApiRefresh: Boolean? = null
-        return loadMoreCombineTransform<List<Label>, GetMessagesResult, Pair<List<Label>, GetMessagesResult>>(
+        return loadMoreCombine(
             observeLabels(userId),
             observeMessagesByLocation(
                 userId = userId,
                 mailboxLocation = location,
                 labelId = labelId
             )
-        ) { messages, labels ->
-            emit(messages to labels)
-        }.loadMoreMap { pair ->
-            val labels = pair.first
-            when (val result = pair.second) {
-                is GetMessagesResult.Success -> {
-                    val shouldResetPosition = isFirstData || hasReceivedFirstApiRefresh == true
-                    isFirstData = false
+        ) { labels, messages -> labels to messages }
+            .loadMoreBuffer()
+            .loadMoreMap { pair ->
+                val labels = pair.first
+                when (val result = pair.second) {
+                    is GetMessagesResult.Success -> {
+                        val shouldResetPosition = isFirstData || hasReceivedFirstApiRefresh == true
+                        isFirstData = false
 
-                    MailboxState.Data(
-                        items = messagesToMailboxItems(result.messages, labels),
-                        isFreshData = hasReceivedFirstApiRefresh != null,
-                        shouldResetPosition = shouldResetPosition
-                    )
-                }
-                is GetMessagesResult.DataRefresh -> {
-                    if (hasReceivedFirstApiRefresh == null) hasReceivedFirstApiRefresh = true
-                    else if (hasReceivedFirstApiRefresh == true) hasReceivedFirstApiRefresh = false
+                        MailboxState.Data(
+                            items = messagesToMailboxItems(result.messages, labels),
+                            isFreshData = hasReceivedFirstApiRefresh != null,
+                            shouldResetPosition = shouldResetPosition
+                        )
+                    }
+                    is GetMessagesResult.DataRefresh -> {
+                        if (hasReceivedFirstApiRefresh == null) hasReceivedFirstApiRefresh = true
+                        else if (hasReceivedFirstApiRefresh == true) hasReceivedFirstApiRefresh = false
 
-                    MailboxState.DataRefresh(
-                        lastFetchedItemsIds = result.lastFetchedMessages.mapNotNull { it.messageId }
-                    )
-                }
-                is GetMessagesResult.Error -> {
-                    hasReceivedFirstApiRefresh = false
+                        MailboxState.DataRefresh(
+                            lastFetchedItemsIds = result.lastFetchedMessages.mapNotNull { it.messageId }
+                        )
+                    }
+                    is GetMessagesResult.Error -> {
+                        hasReceivedFirstApiRefresh = false
 
-                    MailboxState.Error(
-                        error = "GetMessagesResult Error",
-                        throwable = result.throwable,
-                        isOffline = result.isOffline
-                    )
+                        MailboxState.Error(
+                            error = "GetMessagesResult Error",
+                            throwable = result.throwable,
+                            isOffline = result.isOffline
+                        )
+                    }
+                    is GetMessagesResult.Loading ->
+                        MailboxState.Loading
                 }
-                is GetMessagesResult.Loading ->
-                    MailboxState.Loading
             }
-        }
     }
 
     private suspend fun conversationsToMailboxItems(
@@ -436,7 +434,9 @@ internal class MailboxViewModel @Inject constructor(
             allLabels = labels
         )
 
-    private suspend fun messagesToMailboxItems(messages: List<Message>, labelsList: List<Label>?): List<MailboxItemUiModel> {
+    private suspend fun messagesToMailboxItems(
+        messages: List<Message>, labelsList: List<Label>?
+    ): List<MailboxItemUiModel> {
         Timber.v("messagesToMailboxItems size: ${messages.size}")
 
         val labelIds = messages.flatMap { message -> message.allLabelIDs }.distinct().map { LabelId(it) }

domain/src/main/kotlin/ch/protonmail/android/domain/LoadMoreFlowUtils.kt

@@ -19,9 +19,12 @@
 
 package ch.protonmail.android.domain
 
+import kotlinx.coroutines.channels.BufferOverflow
+import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.FlowCollector
 import kotlinx.coroutines.flow.MutableSharedFlow
+import kotlinx.coroutines.flow.buffer
 import kotlinx.coroutines.flow.catch
 import kotlinx.coroutines.flow.combine
 import kotlinx.coroutines.flow.combineTransform
@@ -99,6 +102,12 @@ fun <B, T> Flow<T>.asLoadMoreFlow(
 fun <T> LoadMoreFlow<T>.loadMoreCatch(action: suspend FlowCollector<T>.(Throwable) -> Unit): LoadMoreFlow<T> =
     LoadMoreFlow(catch(action), trigger)
 
+fun <T> LoadMoreFlow<T>.loadMoreBuffer(
+    capacity: Int = Channel.BUFFERED,
+    onBufferOverflow: BufferOverflow = BufferOverflow.SUSPEND
+): LoadMoreFlow<T> =
+    LoadMoreFlow(buffer(capacity, onBufferOverflow), trigger)
+
 /**
  * Same as [combineTransform], but returns a [LoadMoreFlow] instead
  */
@@ -108,6 +117,12 @@ public fun <T1, T2, R> loadMoreCombineTransform(
     transform: suspend FlowCollector<R>.(a: T1, b: T2) -> Unit
 ): LoadMoreFlow<R> = LoadMoreFlow(combineTransform(flow, loadMoreFlow, transform), loadMoreFlow.trigger)
 
+public fun <T1, T2, R> loadMoreCombine(
+    flow: Flow<T1>,
+    loadMoreFlow: LoadMoreFlow<T2>,
+    transform: suspend (a: T1, b: T2) -> R
+): LoadMoreFlow<R> = LoadMoreFlow(combine(flow, loadMoreFlow, transform), loadMoreFlow.trigger)
+
 /**
  * @return [LoadMoreFlow] with nullable [T], emitting one `null` when Flow is started
  *  @see Flow.onStart