[Feature Request] Improved Support for Pull to Refresh

[alexandercasal]

Is your feature request related to a problem? Please describe.
The documentation states "Another good use case for fresh() is when a user wants to pull to refresh." While this is correct, there is a drawback to using fresh() for pull to refresh events: We don't get the Loading or Error states from the StoreResponse. This may mean implementing additional logic around showing the user the loading state or error state in the UI when using both stream(), for the primary flow of data, and fresh() for the pull to refresh event.

Describe the solution you'd like
I created the following extensions to be used in a pull refresh scenario and have found they work well. Since pull to refresh is common, perhaps these would be good to integrate into Store somehow?

/**
 * Use to trigger a network refresh for your Store. An example use-case is for pull to refresh functionality.
 * This is best used in conjunction with `Flow.collectRefreshFlow` to prevent this returned Flow from living
 * on longer than necessary and conflicting with your primary Store Flow.
 */
fun <Key : Any, Output : Any> Store<Key, Output>.refreshFlow(key: Key) = stream(StoreRequest.fresh(key))

/**
 * Helper to observe the loading state of a Store refresh call triggered via `Store.refreshFlow`. This Flow will
 * automatically be cancelled after the refresh is successful or has an error.
 *
 * @param checkLoadingState Lambda providing the current StoreResponse for the refresh (Loading, Error, or Data) allowing
 * you to decide when to show/hide your loading indicator.
 */
suspend fun <T> Flow<StoreResponse<T>>.collectRefreshFlow(checkLoadingState: suspend (StoreResponse<T>) -> Unit) {
    onEach { storeResponse ->
        checkLoadingState(storeResponse)
    }.first { storeResponse ->
        storeResponse.isData() || storeResponse.isError()
    }
}

Examples of the solution in use
Below is an example of how refreshFlow is used alongside the main flow:

fun getUserDataFlow(): Flow<StoreResponse<List<UserData>>> {
   return myStore.stream(StoreRequest.cached(Key("123"), refresh = true))
}

fun refreshUserDataFlow(): Flow<StoreResponse<List<UserData>>> {
   return myStore.refreshFlow(Key("123"))
}

And finally an example of how collectRefreshFlow is used in a ViewModel alongside the main flow:

class MyViewModel {
   val userDataLiveData = userRepository.getUserDataFlow()
      .onEach { storeResponse ->
            setLoadingSpinnerVisibility(storeResponse)
            setEmptyStateVisibility(storeResponse)
       }
       .filterIsInstance<StoreResponse.Data<List<UserData>>>()
       .mapLatest { storeResponse ->
            storeResponse.dataOrNull() ?: emptyList()
       }
       .flowOn(Dispatchers.IO)
       .asLiveData().distinctUntilChanged()
   }

   fun onRetryLoadUserData() = viewModelScope.launch {
      if (networkIsConnected()) {
         userRepository.refreshUserDataFlow()
            .collectRefreshFlow { storeResponse -> setLoadingSpinnerVisibility(storeResponse) }
      }
   }
}

Additional context
I feel these two extensions provide more power/flexibility in the pull to refresh scenario than simply calling the suspending fresh() method. Does it seem like these or something similar could fit into the Store API?

[digitalbuddha]

Does look like a nice api. I'd love to hear other's thoughts. Since this is additive maybe we should have a recipe? This would give us more flexibility with not having to support this api forever while still sharing with others

[yigit]

I actually thought we would write a helper class for this, never get around to finalize it.
The thing i was thinking was more like a helper object that encapsulates the behavior.
Something like this:

class UIRequest(
    val key: Key
) {
   private val requests = ConflatedBroadcastChannel<StoreRequest>(StoreRequest.cached(key = key, refresh = true))
   fun refresh() {
      requests.offer(StoreRequest.fresh(key))
   }
  val stream : Flow<StoreResponse<Key, Value> = refreshRequest.asFlow().flatMapLatest {
       store.stream(it)
  }
}

The problem with the implementation above is that it will NOT reflect any further local change if the network request for refresh fails which is not great. But if desired (usually is), it can be handled w/ a little modification in the steam

class UIRequest(
    val key: Key
) {
   private val requests = ConflatedBroadcastChannel<Boolean>(true)
   fun refresh() {
      requests.offer(true)
   }
  val stream : Flow<StoreResponse<Key, Value> = refreshRequest.asFlow().flatMapLatest {
       store.stream(StoreRequest.cached(key, refresh = it))
  }
}

Now this one has the downside of re-emitting the latest value. Right now, if a fresh request's network request fails, we don't emit any further values. We might either make a flag for that (fallbackToDiskOnError) which might actually make sense (thinking out loud here), or we can change that helper to handle that and re-connect a cached read if a Error is dispatched.

Thoughts ?

[ychescale9]

Right now, if a fresh request's network request fails, we don't emit any further values. We might either make a flag for that (fallbackToDiskOnError)

Was running into this exact issue yesterday when trying to implement an extension that supports conditionally using fresh() / cached() request based on a lambda user passes in.

I’d love to have a fallbackToDiskOnError flag for the fresh() request, although I’d also like the UI to be notified of the failed fetch after displaying the fallback data from cache e.g. showing a snackbar. Perhaps in that case it can emit a Data response with cached data, followed by an Error response with origin = Fetcher, then continue to stream changes from sourceOfTruth? Or can we model this with a single Error response with the data field populated with cache fallback? Neither feels intuitive with the current StoreResponse API though...

[yigit]

Your UI will still know if we add fallback as your client will receive:

StoreResponse.Loading(origin = Fetcher...)
StoreResponse.Error(origin = Fetcher...)
// maybe cache, i'm not sure but should probably also include cache, almost definitely :dancer: 
StoreResponse.Data(origin = Persister) 

do you think it is not enough?

[ychescale9]

It could work for "swipe to refresh" when the UI is already displaying data.

For my case though I want to start streaming with a StoreRequest.fresh(fallbackToDiskOnError = true) at the start. Problem with the Loading, Error, Data sequence is when I receive the Error I don't know whether an incoming Data(origin = Persister will be emitted immediately after it so I don't know whether to show a prominent error state (with a retry button in the middle), or display a transient error message (snackbar) and wait for the upcoming data from persister.

In general I'd like to use a separate stream for handling transient events such as the Error(origin = Fetcher) in this case instead of turning it into a permanent "Error" state which will be cached and re-emitted as the latest state after config change. This is why I don't like the Loading, Error, Data sequence as we'll need to keep track of the previous emission to decide what the right current state should be.

[ychescale9]

My current solution is to stream StoreRequest.cached(key, refresh = true) as the main flow, and use suspend fun Store<Key, Output>.fresh(key: Key) to trigger a fetch for swipe to refresh and rely on the multicaster to emit new data through the main flow.

With this approach the InFlight / Error states need to be managed manually in the ViewModel / StateMachine beyond the initial request, but this is not that big of a deal when you already need to do this anyway to merge multiple streams of data sources into UI states.

[OrhanTozan]

I think introducing a new API for this is a bandage fix. I think its commonly expected that the .fresh() method itself should trigger the Store emitting a StoreResponse.Loading.

[aartikov]

@digitalbuddha @NahroTo
I was surprised when discovered that stream() does not emit Loading and Error for subsequent calls of fresh(). Is not it more handy behaviour?

[ferinagy]

I also agree with @NahroTo and @aartikov that I would expect subsequent loading states to be also emitted. Paraphrasing StreamWithoutSourceOfTruthTest, I would expect:

    @Test
    fun streamWithoutMultipleLoads() = testScope.runBlockingTest {
        val fetcher = FakeFetcher(3 to "three-1", 3 to "three-2")

        val pipeline = StoreBuilder.from(fetcher).scope(testScope).build()

        val twoItemsNoRefresh = async {
            pipeline.stream(
                StoreRequest.cached(3, refresh = false)
            ).take(4).toList()
        }

        delay(1_000) // make sure the async block starts first
        pipeline.fresh(3)

        assertThat(twoItemsNoRefresh.await()).containsExactly(
            StoreResponse.Loading(origin = ResponseOrigin.Fetcher),
            StoreResponse.Data(
                value = "three-1",
                origin = ResponseOrigin.Fetcher
            ),
            StoreResponse.Loading(origin = ResponseOrigin.Fetcher), // this is not present currently
            StoreResponse.Data(
                value = "three-2",
                origin = ResponseOrigin.Fetcher
            )
        )
    }

Currently there is no 2nd Loading in the stream.

PS @aartikov: In my testing (on 4.0.0-alpha07), I am only missing Loading state. So if there is an error, with 2 fetch calls, I see a stream of Loading, Exception, Exception

[aartikov]

@ghus-raba
After some experiments I found out, that second Error is missing when a store has SourceOfTruth.
Here is my test https://gist.github.com/aartikov/34f1413c9bb8d3b548730b668dbfbd5d

@digitalbuddha
Is it an expected behaviour?

[yigit]

hmm that does look like a bug but need to debug further why the second error is not showing, it should show up in the stream.

[yigit]

ok I debugged it, what is going on is that it is a one off fetcher so when the first stream receives an error, it does not have any reason to keep listening for more, it falls back to Source of truth.
We enable piggyback on server responses only if there is no source of truth. I'm not sharp enough to remember why we did it that way.
https://github.com/dropbox/Store/blob/main/store/src/main/java/com/dropbox/android/external/store4/impl/FetcherController.kt#L63

I've tried always enabling it which does fix this test and does not fail any other test but i'm not 100% sure why we did it this way so not feeling very comfortable to make the change. (also, you still won't get another loading)

The right solution might be just creating the helper class that'll do this properly (a stream of refresh triggers).

Something like:

val refreshTrigger = MutableStateFlow(0)
val data = refreshTrigger.flatMap {
    store.stream(...)
}
fun refresh() = refreshTrigger.value += 1

[ferinagy]

Thanks for the tips for the helper @yigit (both #124 (comment) and #124 (comment)). I checked it out and found a few drawbacks in these implementations:

1. It only notifies of the subsequent Loading if it was triggered via refresh(), but not via underlying store
2. It only supports requests with refresh == true -> not too hard to add by changing to fun stream(refresh: Boolean): Flow<StoreResponse<Key, Value>
3. It repeats cached value after refresh, as you mentioned -> I managed to filter the subsequent caches out, but the initial code is a bit messy (probably could be improved)
4. It only supports a single key -> it is probably enough for the pull-to-refresh scenario, but it would be nice to also support stream of states per key (which is already supported by store.stream(...))
5. Not entirely sure, but from the looks of it, I would say, that there might be some issues when subscribing to the stream after the loading started, but before the data is delivered.

That being said, such helper class would be helpful, but I would prefer it to be provided by this library as opposed to trying to write it myself in every project.

---

On the other hand, I still think it would be easier if the stream(...) would also deliver subsequent Loading states and we could just filter them out if not interested instead of this dance with a trigger + flatMap + throwing away the stream every time there is a refresh.

I could imagine to extend the StoreResponse to distinguish reloading and error after data was already delivered:

sealed class StoreResponse<out T> {
    object Loading : StoreResponse<Nothing>
    data class Data(val value: T, origin: ResponseOrigin) : StoreResponse<T>
    data class Error(val error: Throwable, origin: ResponseOrigin) : StoreResponse<T>

    data class LoadingWithPreviousData(val previousData: Data<T>) : StoreResponse<T> 
    data class ErrorWithPreviousData(val error: Throwable, origin: ResponseOrigin, val previousData: Data<T>) : StoreResponse<T>
}

The last 2 states would be useful to eg. show list with data with a loading indicator and show list with snackbar error respectively. If the stream(...) would deliver something like this, we could also show list with loading indicator after subscribing when a request is already in flight, as we would be receiving LoadingWithPreviousData.

[OrhanTozan]

I think @ghus-raba 's solution is nicely intuitive, that's what I found others (including myself) do in the past for repositories. I think it keeps the users away of managing the loading state on their own, and keeping the Store as the source of truth.

Quick note: I would rename Loading to something like InitialLoading.