mobilenativefoundation / store Goto Github PK

Exceptions aren't good. I think the Kotlin community is coalescing around returning a sealed class with error information instead of throwing an exception to indicate that a function has an error. That's what Store does with the StoreResponse class. However, a StoreResponse.Error object will only be returned if the fetcher throws an exception. Now the libraries commonly used by a fetcher, such as Retrofit or Ktor, throw exceptions on error (bad Ktor, no biscuit). But what if someone wants to use a library that returns a sealed class to indicate success or failure?

For example, if MyLibrary returns a MyLibraryResult.Success or a MyLibraryResult.Error, how should the fetcher or Store handle it?

• The app could define a custom exception which the fetcher throws when it receives an error, but we're trying to get away from exceptions. Introducing an exception between two libraries that don't use exceptions doesn't sound right.
• The fetcher could return the MyLibraryResult object (whether or not there was an error), but then Store doesn't know there was a failure. Also the persister has to deal with unwrapping the data while storing it, and handling any errors.
• There could be some direct way for the fetcher to tell Store there was a failure. The fetcher lambda could be an extension function on an interface, perhaps with a reportError() method. If this is called then the fetcher's return value is ignored and it is treated as an error.
• The fetcher returns a StoreResponse.Error object, which Store detects and treats the same as if an exception was thrown.
• The StoreBuilder functions, in addition to the key and output types, has a third type for errors. If the fetcher returns an object of this class, it treats it as an error. For example: StoreBuilder.fromNotFlow<String, Data, MyLibraryResult.Error>(). If the fetcher succeeds, it unwraps the data from the result class and returns it. If it fails, it returns an error object.

The StoreResponse class would need to be modified to handle this new kind of error. Maybe there are two error subclasses, one for exceptions and one for non-exceptions. Maybe StoreResponse.Error can contain either a Throwable or some other error class.

Need to add ktlint for auto formatting. Marked as help wanted as this is a task that someone externally can pick up. If not I'm happy to do it.

Right now, multicast kills the upstream if there are 0 consumers.

It would be nice if it had a mode where it kept the upstream alive as long as the given scope is alive, even if there are no consumers. Alternatively, it could have a delay for killing upstream to achieve this.

This is something we needed in paging. Basically, during rotation, consumer count goes to 0 which means the upstream paging flow will be closed, requiring the new activity to restart it from scratch.
That is waste of work, especially there are intermediate transformations.

There is a hacky workaround where it can keep a collector active in the viewModelScope but that would make the upstream hot.
The desired behavior is to just not cancel it and upstream will eventually suspend as no values will be consumed.

I little bit confuse, when we need to update local data soure if the local data is exists? is it just update when we try to refresh by action (ex: swipe refresh) ?

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?

Describe the bug

When we run the instrumented test in Firebase Test Lab, sometimes App will crash due to the following exception:

kotlin.UninitializedPropertyAccessException: lateinit property collectionJob has not been initialized
at com.dropbox.flow.multicast.SharedFlowProducer.cancel(SharedFlowProducer.kt:93)
at com.dropbox.flow.multicast.ChannelManager$Actor.onClosed(ChannelManager.kt:177)
at com.dropbox.flow.multicast.StoreRealActor.doClose(StoreRealActor.kt:62)
at com.dropbox.flow.multicast.StoreRealActor.access$doClose(StoreRealActor.kt:32)
at com.dropbox.flow.multicast.StoreRealActor$1.invokeSuspend(StoreRealActor.kt:54)
at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:33)
at kotlinx.coroutines.DispatchedTask.run(DispatchedTask.kt:56)
at kotlinx.coroutines.scheduling.CoroutineScheduler.runSafely(CoroutineScheduler.kt:561)
at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.executeTask(CoroutineScheduler.kt:727)
at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.runWorker(CoroutineScheduler.kt:667)
at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.run(CoroutineScheduler.kt:655)

To Reproduce
Not sure how to reproduce it.😥

Smartphone (please complete the following information):

• Store Version [e.g. 4.0.0-alpha02]

If we wanted Store to be the backing data source for paging, we would need to be able to fetch not by key, but by range.
@yigit what do you think? I haven't dug into Paging3 APIs yet (because the only way for me to do it now it to read the source code :)), but feels this is some thing that's missing right now (e.g #63).

Is it possible to execute some custom actions when memory cache become invalidated?

I am trying to integrate Jetpack Paging with Store.
The DataSource required us to call invalidate when we want to stop paging.

Currently, I am using a map to cache the latest result and invalidate old one from the fetcher:

val dataSourceMap = mutableMap<Key, WeakReference<PagedList>>()
val store = StoreBuilder.fromNonFlow { key: Key ->
  val pagedList = createPagedList(key)
  dataSourceMap[key]?.get()?.dataSource?.invalidate()
  dataSourceMap[key] = WeakReferene(pagedList)
  pagedList
}.build()

Maybe it will be convenient if StoreBuilder can provide a onCacheInvalidated method:

val store = StoreBuilder.fromNonFlow { key: Key ->
  createPagedList(key)
}.onCacheInvalidate { pagedList -> 
  pagedList.dataSource.invalidate() 
}.build()

Edit: Use WeakReference

We should make interop with RxJava easier

• Create an RxStoreBuilder which takes singles/observables as sources
• Create an RxStore which returns Singles/Observables
• Create sample using Retrofit/Room/MvRx

Both of the above can be solved with wrappers for current implementations where we then use the coroutine rx conversion libraries for example https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-reactive/kotlinx.coroutines.reactive/org.reactivestreams.-publisher/as-flow.html

As part of the kotlin multi-platform effort #50, let's migrate cache library to use kotlin.time.Duration instead of duration: Long, unit: TimeUnit.

Example:

import kotlin.time.Duration
import kotlin.time.ExperimentalTime
/**
 * Specifies that each entry should be automatically removed from the cache once a fixed duration
 * has elapsed after the entry's creation or the most recent replacement of its value.
 *
 * The [duration] passed in must be positive.
 * 
 * If not set, default value is [Duration.INFINITE].
 */
@ExperimentalTime
fun expireAfterWrite(duration: Duration): Builder

Many long running data loads are able to report progress during the loading. This can allow for a better user experience as users can quickly estimate the time of a download, and take quicker action to check their connectivity if there is an interruption mid-load.

At first blush this seems to require new functionality in the StoreBuilder perhaps a StoreBuilder.fromProgressive(fetcher: (key: Key) -> Flow<Progressive<Output>>).

As an InputStream is most commonly buffered into a FileOutputStream to avoid loading the entire resource into memory, it would be the responsibility of the fetcher to write to disk.

It could be that this use case is beyond the scope of the project, in which case I humbly ask that this be advertised in the README, as it took quite a while to understand the current abilities of this library.

Thanks.

When I run the sample app, there are 3 different apps installed for each activity (RedditActivity, RoomActivity and StreamActivity). Is there any reason to do that?

right now we have a multicast#close function that closes the actor, making it useless.
After that point, any new downstream flow will receive closed channel exception.
We should instead send a more explicit exception so that developer can understand what happened.

private fun sensorMapping(sensor: SensorApi.Dto.SensorRecentResponse) =
        sensor.map()

    val store =
        StoreBuilder
            .fromNonFlow<StoreKey.SensorsKey, List<SensorData>> {
                remoteDataSource.getSensors().body()?.map(::sensorMapping)!!
            }
            .persister(
                reader = localDataSource::getSensors,
                writer = localDataSource::saveSensors
            ).cachePolicy(
                MemoryPolicy.builder()
                    .setMemorySize(100)
                    .setExpireAfterAccess(8)
                    .setExpireAfterTimeUnit(TimeUnit.DAYS)
                    .build()
            ).build()

We are getting the body that has our data as I've confirmed in a breakpoint but store isn't actually doing anything with it, so it isn't an issue with the endpoint. We are collecting the flow, but we set a breakpoint in the line: remoteDataSource.getSensors().body()?.map(::sensorMapping)!! and it doesn't even reach that breakpoint.

It used to reach that breakpoint when it had this as the body in nonFlow:

val sensors = mutableListOf()
remoteDataSource.getSensors().body()?.forEach {
sensors.add(it.map())
}
sensors

But the problem with this was that it never reached the line sensors where we returned the list of sensors

Memory and disk caches can be out of sync. One scenario where this happens is when a store has 0 active collectors and a new value is saved to your source of truth. As a result we always emit disk and memory cache values one after the other whenever a cache value is requested. This is not ideal for consumers that don't have diffing. I'd like to add a store configuration that compares disk to memory and if they are the same then don't update and emit the memory value.

It'd be good to start tracking the binary compatibility of our APIs once we're stable.

Kotlin released a new binary-compatibility-validator plugin which seems fairly easy to integrate and use. I'll create a draft PR for us to explore and decide if this is something we'd like to use.

It seems that even if we set refresh flag to false in StoreRequest.cached, we still get the latest data from the fetcher.
Is this the correct behavior?

Test code:

runBlocking {
    // Store with cache, without persister
    val store = StoreBuilder.fromNonFlow { key: Int -> key }.build()

    // Cache data
    store.fresh(1)

    // Stream data
    store.stream(StoreRequest.cached(1, refresh = false)).collect {
        println(it.toString())
    }
}

Output:

Data(value=1, origin=Cache)
Loading(origin=Fetcher)
Data(value=1, origin=Fetcher)

The dependency - implementation 'com.dropbox.mobile.store4:4.0.0-SNAPSHOT' failing to resolve. it is included in build.gradle(Module: app)
I need help fixing it.
@yigit @digitalbuddha

If network happens to be a lot faster than disk, it might actually write the data into disk before we start observing which will make SourceOfTruth consider it a disk value rather than fetcher.
We should block the network until disk flow is established. This does not mean we'll read data from disk as it will wait for a lock before continuing. This only ensures the ordering so that it can know whether a network value is dispatched for the same request or for an earlier one.

Perhaps Travis can look at the commit message and if it starts with "Preparing for release", then it would run ./gradlew clean uploadArchives.

• Also need to store the Sonatype user name and password in Travis

After reading #73 and the RealStore implementation, I realized it's effectively useless to set expireAfterAccess or expireAfterWrite on the MemoryPolicy when a sourceOfTruth is provided to the store, as the disk value will be immediately backfilled to the cache after cache expiration. From the user's perspective, they'll get the data from the disk instead of cache (slightly slower) when cache has expired but the fetcher won't be triggered (unless they bypass Store and manually delete the entry from disk).

This behavior is inconsistent to when sourceOfTruth == null, in which case cache expiry will trigger the fetcher (thanks to piggybackOnly).

Should we try to make these 2 branches (with or without sourceOfTruth) behave consistently in terms of the effect of cache expiry?

I remember the old Store had a networkBeforeStale() API for triggering the fetcher if the disk entry is stale. Does it make sense to support this (on both memory cache and sourceOfTruth?)?

I understand that cache expiry and disk entry becoming stale could be interpreted as different concepts so unifying the behavior with or without sourceOfTruth might cause other confusions...

I'm using Store with a persister implementation that uses android's fun Context.getCacheDir() to save media files to disk. Is there a way to control the maximum size or maximum number of these cached files?

How can I get timestamp when data last time was accessed or written in the store?

Hi :)

Do you plan to add some support to RxJava in near future?

Is your feature request related to a problem? Please describe.
Store::get and Store::fresh allow easy retrieval of a value from the Store however it would be nice to have an Rx equivalent to bridge the use case where a project has elected to use RxJava over Coroutines.

Describe the solution you'd like
Extension functions on Store that return the idiomatic RxJava type but otherwise operate in the same way as Store::get and Store::fresh.

Describe alternatives you've considered
Writing this manually using existing Rx operators such as:

// To recreate Store::get(key)
store.observe(StoreRequest(key, refresh = false))
        .filter { it !is StoreResponse.Loading }
        .firstOrError()
        .map { it.requireData() }     

Additional context
I've forked the project and created a branch for which I am more than happy to submit a PR for but I wanted to discuss this first if this was something you'd be open to. Also I wasn't sure what to do about tests so some guidance there would be appreciated.

mezpahlan@bba706b

There might be an edge case in channel manager where it might idle forever.
have not tried but we should check: (buffer = 0)

• add a downstream that is suspending on collect
• dispatch value from upstream
• add another downstream
• remove first downstream

I believe it will be broken such that the first value will never be acknowledged but the second downstream will just await forever.

Seems like we need some more tracking on message acks

Most tests still use junit assertions, it would be nice to covert tests to instead use Truth library

The old Store had a clear() API which purges all entries in memory and disk. The new store currently only has suspend fun clear(key: Key).

Should we add suspend fun clearAll()?

This may be trivial, but I was unsure whether the argument to fun setMemorySize(maxSize: Long) referred to bytes, KB or MB. Could this be clarified in documentation?

I've noticed this while trying to repro issue #86 .

2020-02-03 21:07:37.721 11904-11904/com.dropbox.stor4.bug E/AndroidRuntime: FATAL EXCEPTION: main
    Process: com.dropbox.stor4.bug, PID: 11904
    java.util.ConcurrentModificationException
        at java.util.LinkedHashMap$LinkedHashIterator.nextNode(LinkedHashMap.java:757)
        at java.util.LinkedHashMap$LinkedKeyIterator.next(LinkedHashMap.java:780)
        at kotlin.collections.CollectionsKt___CollectionsKt.first(_Collections.kt:185)
        at com.dropbox.android.external.cache4.RealCache.evictEntries(RealCache.kt:218)
        at com.dropbox.android.external.cache4.RealCache.put(RealCache.kt:147)
        at com.dropbox.android.external.store4.impl.RealStore$stream$2.invokeSuspend(RealStore.kt:114)
        at com.dropbox.android.external.store4.impl.RealStore$stream$2.invoke(Unknown Source:10)
        at kotlinx.coroutines.flow.FlowKt__TransformKt$onEach$$inlined$unsafeTransform$1$2.emit(Collect.kt:138)
        at kotlinx.coroutines.flow.internal.SafeCollector.emit(SafeCollector.kt:33)
        at kotlinx.coroutines.flow.internal.SafeCollector.emit(SafeCollector.kt:33)
        at kotlinx.coroutines.flow.internal.SafeCollector.emit(SafeCollector.kt:33)
        at kotlinx.coroutines.flow.FlowKt__ErrorsKt$catchImpl$$inlined$collect$1.emit(Collect.kt:138)
        at kotlinx.coroutines.flow.internal.SafeCollector.emit(SafeCollector.kt:33)
        at com.dropbox.flow.multicast.Multicaster$newDownstream$2$invokeSuspend$$inlined$transform$1$1.emit(Collect.kt:138)
        at kotlinx.coroutines.flow.FlowKt__ChannelsKt.emitAll(Channels.kt:56)
        at kotlinx.coroutines.flow.FlowKt.emitAll(Unknown Source:1)
        at kotlinx.coroutines.flow.FlowKt__ChannelsKt$emitAll$1.invokeSuspend(Unknown Source:10)
        at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:33)
        at kotlinx.coroutines.DispatchedTask.run(DispatchedTask.kt:56)
        at kotlinx.coroutines.scheduling.CoroutineScheduler.runSafely(CoroutineScheduler.kt:561)
        at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.executeTask(CoroutineScheduler.kt:727)
        at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.runWorker(CoroutineScheduler.kt:667)
        at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.run(CoroutineScheduler.kt:655)

https://github.com/atonamy/store4-issue

To reproduce, checkout the app and hit refresh couple of times and eventually happens. Sorry i know it is not great repro but i'm focusing on the other bug right now so wanted to report this w/o being side-tracked.
.

I wrote a test using both alpha and snapshot and both fail by not emitting the data. Is there something that needs to be adjusted? Opening issue for tracking I will continue to look tomorrow:

class PadStoreTest {
    private val testScope = TestCoroutineScope()

    lateinit var padStore: Store<String, String>
    @Before
    fun setup() {
        /*
         StoreBuilder
            .fromNonFlow<String, String> {
                apiService.getPadIdAndFileObjIdFromCloudDocId(it).localPadId
            }
            .nonFlowingPersister(
                writer= {key,value->keyValueStore.update(StoreKey.StringValue(key), value)},
                reader = {key-> keyValueStore.getSuspendedString(StoreKey.StringValue(key))})
            .build()
         */
        padStore = PadStoreModule.providePadStore(fakePadApi, fakeSuspendKeyValueStore)
    }

    @Test
    fun `Given a PadStore WHEN data is fetched THEN first load from network AND then load from cache`() {
        testScope.runBlockingTest {
           assertThat(padStore.stream(StoreRequest.fresh("5")))
               .emitsExactly(
                   StoreResponse.Loading(
                       origin = ResponseOrigin.Fetcher
                   ),
                   StoreResponse.Data(
                       value = "55",
                       origin = ResponseOrigin.Fetcher
                   )
               )

        }
    }

}

private val fakePadApi: CloudToPadIDApiService = object : CloudToPadIDApiService {
    override suspend fun getPadIdAndFileObjIdFromCloudDocId(cloudDocId: String): GetCloudDocInfoResponse {
        return GetCloudDocInfoResponse(cloudDocId, cloudDocId + cloudDocId, "")
    }
}

private val fakeSuspendKeyValueStore: SuspendingKeyValueStore = object : SuspendingKeyValueStore {
    val fakeDisk: MutableMap<StoreKey, String> = mutableMapOf(StoreKey.StringValue("5") to "55")

    override suspend fun update(storeKey: StoreKey, value: String) {
        fakeDisk[storeKey] = value
    }

    override suspend fun getSuspendedString(storeKey: StoreKey): String? {
        return fakeDisk[storeKey]
    }
}```

Flow didn't exactly emit expected items
missing (1): Data(value=55, origin=Fetcher)

expected : [Loading(origin=Fetcher), Data(value=55, origin=Fetcher)]
but was : [Loading(origin=Fetcher)]

at com.dropbox.paper.pifs.PadStoreTest$Given a PadStore WHEN data is fetched THEN first load from network AND then load from cache$1.invokeSuspend(PadStoreTest.kt:39)
at com.dropbox.paper.pifs.PadStoreTest$Given a PadStore WHEN data is fetched THEN first load from network AND then load from cache$1.invoke(PadStoreTest.kt)
at kotlinx.coroutines.test.TestBuildersKt$runBlockingTest$deferred$1.invokeSuspend(TestBuilders.kt:50)
at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:33)
at kotlinx.coroutines.DispatchedTask.run(DispatchedTask.kt:56)
at kotlinx.coroutines.test.TestCoroutineDispatcher.dispatch(TestCoroutineDispatcher.kt:50)
at kotlinx.coroutines.DispatchedContinuationKt.resumeCancellableWith(DispatchedContinuation.kt:271)
at kotlinx.coroutines.intrinsics.CancellableKt.startCoroutineCancellable(Cancellable.kt:26)
at kotlinx.coroutines.CoroutineStart.invoke(CoroutineStart.kt:109)
at kotlinx.coroutines.AbstractCoroutine.start(AbstractCoroutine.kt:158)
at kotlinx.coroutines.BuildersKt__Builders_commonKt.async(Builders.common.kt:89)
at kotlinx.coroutines.BuildersKt.async(Unknown Source)
at kotlinx.coroutines.BuildersKt__Builders_commonKt.async$default(Builders.common.kt:82)
at kotlinx.coroutines.BuildersKt.async$default(Unknown Source)
at kotlinx.coroutines.test.TestBuildersKt.runBlockingTest(TestBuilders.kt:49)
at kotlinx.coroutines.test.TestBuildersKt.runBlockingTest(TestBuilders.kt:72)
at com.dropbox.paper.pifs.PadStoreTest.Given a PadStore WHEN data is fetched THEN first load from network AND then load from cache(PadStoreTest.kt:37)

Currently Store only supports reading data, it does not provide any support for writing local updates back up to the server.

We would like to add write support to Store that considers offline updates as a first class citizen requirement.

In order to achieve the above we will need to deal to introduce support for things like:

• pending operations
• conflict resolution
• updating the remote
• etc.

Some discussion has been started on this subject at https://paper.dropbox.com/doc/Offline-First-Thoughts-Doc--Ap0nQdYi5xHxXTFl3K~WSGmoAg-JCrUclQot6HBhydtFYxNY

Migrating to Store4 did not migrate the file system module and file system persister to work with the coroutine store4. This issue is to take the file system persister and make it work with source of truth abstraction

Great people are reporting great bugs. Let's make it easier on us and then by setting up an intake template for issues.

Ottomh we should ask for:
Store version
Minimal repro

Let's review this simple example

  fun fetchFromCollection(keys: Set<MySerializedType>){
        someNonblockingScope.launch {
            for(key in keys) {
                myStore.steam(StoreRequest.cached(key, true)).collect {
                    if (it.origin == ResponseOrigin.Fetcher && it !is StoreResponse.Loading)
                        Log.d("CHECKING TRIGGER", "$key triggered")
                }
            }
        }
    }

the expected outcome should be that CHECKING TRIGGER will trigger once for each unique key even if fetchFromCollection executed several times (each time always unique keys)?

If I execute fetchFromCollection with one unique set of keys it will run as expected, but if I execute second time fetchFromCollection with another different set of keys it will trigger CHECKING TRIGGER condition more than once(new set of unique keys).

What I miss? Why it doesn't work as expected?

And then if execute fetchFromCollection three, four times and so on each time with new set of keys stream will just hang with Loading state forever.

I pushed this example project demonstrating the issue in full scale.

Currently we are using a forked version of Guava Cache but no longer need a "Loading Cache". Since our api is much smaller, we should rewrite the Cache. Cache is our last Java module, a rewrite opens up multiplatform support

We need a cache with the following features

• 100% kotlin
• Allows TTL & Size expiration

we should go through existing public APIs and make sure we are not opening up stuff that we don't intend do, make sure documentation is there, understandable etc.

Porbably an easy way to do this is to build docs and read :)

Now that we are getting external contributions, I think it would be beneficial to have some sort of code coverage report in the prs. @changusmc & @wclausen do you have any thoughts on what we should use?

As I understand it, there is no way to specify the maximum number of items, total size, expiry time, purging policy etc. of locally persisted data items. This could lead to the unbounded growth of cached data and force the system to purge locally cached data or issue exceptions.

The responsibility is for the client to implement local cache purging, perhaps in the writer method, or as a background service.

The use case for using Store as a image/video caching library would seem to require such functionality. Is this beyond the scope of the project?

Thanks.

It would be supercool if Store(4)'s multicast implementation could get modularized, so it would be fetchable separately in projects that do not use Store(4), but that need a RxJava.share() alternative in the coroutines world.

Thanks again for the fantastic talk at KotlinConf 2019!

Once #49 lands we will be 100% kotlin. Next, we should create kotlin multiplatform targets and explore current solutions for handling flow/suspend within native/js

Hi, I'm thinking if it is somehow possible to use paged content with the Store. I want to load feed with the newest data and when the user reaches the end of the feed then I want to load older data and so on.

I know that caching of such results can be hard since items can added, or removed, or moved. So I don't know if it is possible. One solution that I figured out would be to cache only the first page and then load others but I don't know how to implement this with the Store.

Can you help me with that?

Thank you for your response

I use inline class to strongly-type string classes and would like to use these as Key and Output in the Store. When implementing a persister, this lead to an exception e.g:

Exception in thread "DefaultDispatcher-worker-2 @coroutine#7" java.lang.ClassCastException: com.example.app.KeyPath cannot be cast to java.lang.String

I'm not sure if this is a user error, an unsupported use-case or a bug.

Here is a small reproduction:

package com.example.app

import com.dropbox.android.external.store4.MemoryPolicy
import com.dropbox.android.external.store4.StoreBuilder
import com.dropbox.android.external.store4.get
import kotlinx.coroutines.ExperimentalCoroutinesApi
import kotlinx.coroutines.FlowPreview
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.runBlocking
import org.junit.jupiter.api.Test

inline class KeyPath(val string: String)
inline class LocalFilePath(val string: String)

@ExperimentalCoroutinesApi
@FlowPreview
class MediaStoreTest {
    @Test
    fun testStoreInlineClassKey() {
        val localDirectory = "/data/user/0/com.example.app/cache/"
        val store = StoreBuilder.from { keyPath: KeyPath ->
            flow { emit(LocalFilePath("$localDirectory${keyPath.string}")) }
        }.persister(reader = {
            val local: LocalFilePath? = null
            flow { emit(local) }
        }, writer = { key, output ->
        }, delete = {
        })
            .build()
        runBlocking {
            store.get(KeyPath("hello.txt"))
        }
    }
}

Stack trace:

Exception in thread "DefaultDispatcher-worker-2 @coroutine#7" java.lang.ClassCastException: com.example.app.KeyPath cannot be cast to java.lang.String
	at com.example.app.MediaStoreTest$testStoreInlineClassKey$store$3.invoke(MediaStoreTest.kt)
	at com.dropbox.android.external.store4.impl.PersistentSourceOfTruth.write(SourceOfTruth.kt:44)
	at com.dropbox.android.external.store4.impl.SourceOfTruthWithBarrier.write(SourceOfTruthWithBarrier.kt:95)
	at com.dropbox.android.external.store4.impl.FetcherController$fetchers$1$3.invokeSuspend(FetcherController.kt:77)
	at com.dropbox.android.external.store4.impl.FetcherController$fetchers$1$3.invoke(FetcherController.kt)
	at com.dropbox.flow.multicast.ChannelManager$Actor.doDispatchValue(ChannelManager.kt:184)
	at com.dropbox.flow.multicast.ChannelManager$Actor.handle(ChannelManager.kt:118)
	at com.dropbox.flow.multicast.ChannelManager$Actor.handle(ChannelManager.kt:86)
	at com.dropbox.flow.multicast.StoreRealActor$1.invokeSuspend(StoreRealActor.kt:50)
	at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:33)
	at kotlinx.coroutines.DispatchedTask.run(DispatchedTask.kt:56)
	at kotlinx.coroutines.scheduling.CoroutineScheduler.runSafely(CoroutineScheduler.kt:561)
	at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.executeTask(CoroutineScheduler.kt:727)
	at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.runWorker(CoroutineScheduler.kt:667)
	at kotlinx.coroutines.scheduling.CoroutineScheduler$Worker.run(CoroutineScheduler.kt:655)

Thanks!

Multicaster#create returns a flow which uses a channel outside itself which would mean a second collection on that flow will cause issues as the channel is already in use, maybe closed etc.

We've not discovered this as we always create() in tests.

Will create a test and a fix.

@changusmc not sure what we need to do to get this onto Dropboxes Travis instance. Let's touch base Monday

It seems like ktlint keeps triping me up because it doesn't support running as part of the workflow (I make a small change, forget to fix some spacing then get a failure). I'd like to try out https://github.com/facebookincubator/ktfmt.
I know the maintainers and would love to give it a try

It says on the front page that the stream is the "primary function" provided by a store instance.

Do I have it right that the main benefit is receiving a StoreResponse that contains a possible 3 different values making reacting to it easy?

How would opening multiple flows work? If a user clicks on a button and StoreRequest.fresh() was called, does that create a brand new flow every single time the function is called? When does the flow stop the collection process?