[BUG] Calling fresh() on Store does not do anything

[dbobrzyk]

Sample app Issue is recreated in sample app: https://github.com/dbobrzyk/StoreIssueSample

Describe the bug

When creating Store from the flow, data loads perfectly on init. But when requested to refresh the data (from button or swipe refresh) calling fresh() on store is doing nothing (nothing like the inside of the function was empty - just void).

Expected behaviour would be to get the new fresh data from repository and pass it to the VM.

To Reproduce

1. Launch the app
2. Wait for the data to load
3. Click Refresh button
4. Request to Fake Service is never made

Expected behavior After clicking the refresh button there should be request made to the Fake Service and new data should be loaded.

Screenshots Instead of screenshoot I created sample app.

Smartphone (please complete the following information):

• Device: Pixel 5 (emulator)
• OS: Android 11 (Api 30)
• Store Version: 4.0.4-KT15

Additional context

After calling SampleRepository.kt line 42 inside of the function in line 23 is never called.

[digitalbuddha]

Hihi I think we have a test that covers this case, check this test or the one right after. I think you should try to figure out what is different from your code. It’s hard for me to tell since it is a larger example with lots going on https://github.com/dropbox/Store/blob/e571b377e5c1c9b7200998d91f2415dcbd507756/store/src/test/java/com/dropbox/android/external/store4/impl/FlowStoreTest.kt#L166

[alex-burghuber]

I have this issue as well and I think the problem is that a store with Fetcher.ofFlow doesn't react to a .fresh(). It seems this is the case in the provided sample app and the test you have linked only tests a normal Fetcher.of for .fresh()

[digitalbuddha]

Hi sorry. I don't quite understand. Could you reword your comment please

[alex-burghuber]

Is it clear now? I accidentally typed Fetcher.ofFlow instead of Fetcher.of in the last sentence

[digitalbuddha]

Aha that explains it. I'll take a look in am and write a failing test

[vovkab]

Seeing similar issue, when using rxjava:

1. store.freshSingle() would fail with an exception Expected at least one element
2. store.observe() would complete without emitting any value
3. and most critical issue is that when it happens it will disconnect any active observer, they all receive onComplete() event, thus breaking all chains, even though per contract they should stay subscribed.

Additionally I've noticed that even though we are passing rxjava scheduler it still, in some cases, going to run Store.stream on a dispatcher thread.

[digitalbuddha]

Hi folks, could you tell me what I am doing wrong with my test case? As in could you make this test fail please, I think I am not quite understanding what the issue is and would love a failing test case.

@FlowPreview
@ExperimentalCoroutinesApi
@RunWith(JUnit4::class)
class FlowStoreTest {
    private val testScope = TestCoroutineScope()

    private fun <Key : Any, Output : Any> StoreBuilder<Key, Output>.buildWithTestScope() =
        scope(testScope).build()
    
    @Test
    fun fetcherOfFlowOnFresh() = testScope.runBlockingTest {
        var requestNumber = 1
        val textFlow =  MutableStateFlow<Any>(1)

        val fetcher = Fetcher.ofFlow<Any, Any> {
            flow {
                emitAll(textFlow.map {  requestNumber++ })
            }
        }

        val store = StoreBuilder.from(fetcher = fetcher).buildWithTestScope()
        assertThat(store.stream(StoreRequest.fresh("KEY"))).emitsExactly(
            Loading(
                origin = ResponseOrigin.Fetcher
            ),
            Data(
                value = 1,
                origin = ResponseOrigin.Fetcher
            ),
        )

        assertThat(store.fresh("KEY")).isEqualTo(2)

        assertThat(store.stream(StoreRequest.fresh("KEY"))).emitsExactly(
            Loading(
                origin = ResponseOrigin.Fetcher
            ),
            Data(
                value = 3,
                origin = ResponseOrigin.Fetcher
            ),
        )
    }

[vovkab]

The issue with the wrong thread (dispatcher) could be reproduced by removing await() from RxSingleStoreExtensionsTest.

For example to confirm current thread:

store.getSingle(3)
    .doOnSuccess { println("getSingle thread=" + Thread.currentThread().name) }
    ...

Would print that we are on a wrong thread:

getSingle thread=DefaultDispatcher-worker-2 @coroutine#1

Similar test but with observe:

store.observe(StoreRequest.cached(3, false))
    .doOnNext { println("observe thread=" + Thread.currentThread().name) }
    .test()

Shows correct thread and works fine without using await:

observe thread=Test worker

p.s. maybe it should be reported as a separate issue?

[muthuraj57]

@digitalbuddha I'm also facing a similar issue. Here is a test case that can reproduce the issue. This test case won't fail, it won't even complete, but we can see the issue from the println logs.

@Test
    fun fetcherOfFlowOnFresh() = testScope.runTest {

        val firstFetcher = Fetcher.ofFlow<Any, Any> {
            flow {
                emit(1)
            }
        }
        val firstStore = StoreBuilder.from(fetcher = firstFetcher).buildWithTestScope()

        var anotherRequestNumber = 1
        val fetcher = Fetcher.ofFlow<Any, Any> {
            println("Fetcher called")
            firstStore.stream(StoreRequest.fresh("KEY"))
                .map { anotherRequestNumber++ }
        }

        val store = StoreBuilder.from(fetcher = fetcher).buildWithTestScope()

        launch {
            store.stream(StoreRequest.fresh("KEY"))
                .onEach {
                    println("First stream response: $it")
                }.launchIn(this)
        }

       //Delay to make sure all the events are emitted from the above flow.
        delay(1000)
        store.fresh("KEY")
    }

Here the first flow is still active while trying to refresh the same store. This is the output printed for this.

First stream response: Loading(origin=Fetcher)
Fetcher called
First stream response: Data(value=1, origin=Fetcher)
First stream response: Data(value=2, origin=Fetcher)

Here we can see that the Fetcher is not called again. The fresh call will never be complete here.

I modified the above to see the issue more clearly.

@Test
    fun fetcherOfFlowOnFresh() = testScope.runTest {

        val firstFetcher = Fetcher.ofFlow<Any, Any> {
            flow {
                emit(1)
            }
        }
        val firstStore = StoreBuilder.from(fetcher = firstFetcher).buildWithTestScope()

        var anotherRequestNumber = 1
        val fetcher = Fetcher.ofFlow<Any, Any> {
            println("Fetcher called")
            firstStore.stream(StoreRequest.fresh("KEY"))
                .map { anotherRequestNumber++ }
        }

        val store = StoreBuilder.from(fetcher = fetcher).buildWithTestScope()

        launch {
            store.stream(StoreRequest.fresh("KEY"))
                .onEach {
                    println("First stream response: $it")
                }.launchIn(this)
        }

        //Delay to make sure all the events are emitted from the above flow.
        delay(1000)
        launch {
            store.stream(StoreRequest.fresh("KEY"))
                .onEach {
                    println("Second stream response: $it")
                }.launchIn(this)
        }
    }

Now the output is,

First stream response: Loading(origin=Fetcher)
Fetcher called
First stream response: Data(value=1, origin=Fetcher)
First stream response: Data(value=2, origin=Fetcher)
Second stream response: Loading(origin=Fetcher)

Here we can see that the Loading event is emitted for the second fresh call, but the fetcher is not called again and thus no data is emitted.

Basically, when another flow for the same store is active, fresh calls are not working.

---

If I cancel the first flow before calling fresh again, the output is printed correctly on unit tests.

@Test
    fun fetcherOfFlowOnFresh() = testScope.runTest {

        val firstFetcher = Fetcher.ofFlow<Any, Any> {
            flow {
                emit(1)
            }
        }
        val firstStore = StoreBuilder.from(fetcher = firstFetcher).buildWithTestScope()

        var anotherRequestNumber = 1
        val fetcher = Fetcher.ofFlow<Any, Any> {
            println("Fetcher called")
            firstStore.stream(StoreRequest.fresh("KEY"))
                .map { anotherRequestNumber++ }
        }

        val store = StoreBuilder.from(fetcher = fetcher).buildWithTestScope()

        val job = launch {
            store.stream(StoreRequest.fresh("KEY"))
                .onEach {
                    println("First stream response: $it")
                }.launchIn(this)
        }

        //Delay to make sure all the events are emitted from the above flow.
        delay(1000)
        launch {
            job.cancel()
            store.stream(StoreRequest.fresh("KEY"))
                .onEach {
                    println("Second stream response: $it")
                }.launchIn(this)
        }
    }

First stream response: Loading(origin=Fetcher)
Fetcher called
First stream response: Data(value=1, origin=Fetcher)
First stream response: Data(value=2, origin=Fetcher)
Second stream response: Loading(origin=Fetcher)
Fetcher called
Second stream response: Data(value=3, origin=Fetcher)
Second stream response: Data(value=4, origin=Fetcher)

But in my real app (Android) where I have a similar setup, even canceling the previous flow doesn't make the fetcher being called again.

Store version used: 5.0.0-alpha03

[muthuraj57]

But in my real app (Android) where I have a similar setup, even canceling the previous flow doesn't make the fetcher being called again.

An update regarding this. I actually subscribed to the same store's stream in two different places and I didn't cancel one of them while trying to call fetch again. When I cancel all of those before trying to do fetch again, fetcher is called properly.

So the issue is, when there is an active subscriber to the store's flow somewhere, fetcher is not being called when calling fetch/cache with refresh true.

[matt-ramotar]

Thanks @muthuraj57 👍🏽 Will take a look

[matt-ramotar]

Hey @muthuraj57 - Sorry to be slow. Spent a few hours on this today. @digitalbuddha and I are meeting tomo and can sync then. In the interim can you give more context on your use case? I can repro with your test as written. However AFAICT rewriting your test to map outside of fetcher gives the expected result.

[muthuraj57]

Thanks for looking into this. My use case is, I have a Store let's call it DataFetchStore in 2 ViewModels. Both ViewModels have the same lifecycle owner. I call stream on both stores in both ViewModels from the init block, so two flow collectors for the same store are active at the same time.

Now at some point, I need to refresh the data from the first ViewModel, so I cancel the previous collect job and call stream again, and this doesn't emit anything other than the Loading state. This is my issue.

If I cancel the collect job from the other ViewModel too before calling stream again, the fetcher is being called and everything works perfectly fine.

[matt-ramotar]

That part makes sense to me 👍🏽. What is the use case for having a fetcher stream another store?

[muthuraj57]

Let's say in a list and detail screen, I combine a few stores and fetch the data to populate the list screen.

Now in the detail screen, I need to get the same data to display the details. To not duplicate the combine operator and all the code inside that, I moved those into a new store and gave it inside fetcher.

So I'll use this new store to fetch data from the list screen and use the same store in the detail screen to get the cached data.

[matt-ramotar]

Can you share a sample with me? Pipeline use case is something we have thought about. My email is [email protected] if you can't share publicly. I'd like to read closely to better understand real use case