From 1da2ca6d1f98d555f792e0d989a1d4e907bd659d Mon Sep 17 00:00:00 2001 From: Tangem Date: Tue, 12 Mar 2024 16:18:53 +0000 Subject: [PATCH] Updated on 2026-08-14 --- .../local/datastore/FileDataStore.kt | 17 +++++-- .../data/common/cache/DefaultCacheRegistry.kt | 47 ++++++++++++------- .../repository/DefaultNetworksRepository.kt | 42 +++++++++++------ .../repository/DefaultQuotesRepository.kt | 25 ++++++---- 4 files changed, 87 insertions(+), 44 deletions(-) diff --git a/core/datasource/src/main/java/com/tangem/datasource/local/datastore/FileDataStore.kt b/core/datasource/src/main/java/com/tangem/datasource/local/datastore/FileDataStore.kt index e9b2317f35..f67f25584f 100644 --- a/core/datasource/src/main/java/com/tangem/datasource/local/datastore/FileDataStore.kt +++ b/core/datasource/src/main/java/com/tangem/datasource/local/datastore/FileDataStore.kt @@ -8,6 +8,8 @@ import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.distinctUntilChanged import kotlinx.coroutines.flow.filterNotNull import kotlinx.coroutines.flow.map +import kotlinx.coroutines.sync.Mutex +import kotlinx.coroutines.sync.withLock import timber.log.Timber @Deprecated("Use shared preferences data store instead") @@ -16,6 +18,7 @@ internal class FileDataStore( private val adapter: JsonAdapter, ) : StringKeyDataStore { + private val mutex = Mutex() private val writeTrigger = Trigger() override suspend fun isEmpty(): Boolean { val e = NotImplementedError("`isEmpty()` function not implemented for `FileDataStore`") @@ -28,7 +31,11 @@ internal class FileDataStore( override fun get(key: String): Flow { return writeTrigger - .map { getInternal(key) } + .map { + mutex.withLock { + getInternal(key) + } + } .filterNotNull() .distinctUntilChanged() } @@ -53,10 +60,12 @@ internal class FileDataStore( override suspend fun store(key: String, value: Value) { try { - val json = adapter.toJson(value) + mutex.withLock { + val json = adapter.toJson(value) - fileReader.rewriteFile(json, key) - writeTrigger.trigger() + fileReader.rewriteFile(json, key) + writeTrigger.trigger() + } } catch (e: Throwable) { Timber.e(e, "Unable to write file: $key") } diff --git a/data/common/src/main/kotlin/com/tangem/data/common/cache/DefaultCacheRegistry.kt b/data/common/src/main/kotlin/com/tangem/data/common/cache/DefaultCacheRegistry.kt index 6d6f19cd0c..d0938c6718 100644 --- a/data/common/src/main/kotlin/com/tangem/data/common/cache/DefaultCacheRegistry.kt +++ b/data/common/src/main/kotlin/com/tangem/data/common/cache/DefaultCacheRegistry.kt @@ -3,15 +3,21 @@ package com.tangem.data.common.cache import com.tangem.datasource.local.cache.CacheKeysStore import com.tangem.datasource.local.cache.model.CacheKey import kotlinx.coroutines.NonCancellable +import kotlinx.coroutines.sync.Mutex +import kotlinx.coroutines.sync.withLock import kotlinx.coroutines.withContext import org.joda.time.Duration import org.joda.time.LocalDateTime import timber.log.Timber +import java.util.concurrent.ConcurrentHashMap internal class DefaultCacheRegistry( private val cacheKeysStore: CacheKeysStore, ) : CacheRegistry { + private val mutex = Mutex() + private val mutexes = ConcurrentHashMap() + override suspend fun isExpired(key: String): Boolean { val cacheKey = cacheKeysStore.getSyncOrNull(key) ?: return true @@ -41,27 +47,36 @@ internal class DefaultCacheRegistry( expireIn: Duration, block: suspend () -> Unit, ) { - val isExpired = isExpired(key) || skipCache - if (!isExpired) return + // use a separate mutexForKey for each key to avoid multiple calls block() to the same key + // also used mutex to safe create mutexForKey, otherwise it can lead to multiple calls for the same key + val mutexForKey = mutex.withLock { + mutexes.getOrPut(key) { Mutex() } + } + mutexForKey.withLock { + val isExpired = isExpired(key) || skipCache + if (!isExpired) { + return + } - try { - Timber.d("Invoke the action associated with the cache key: $key") + try { + Timber.d("Invoke the action associated with the cache key: $key") - cacheKeysStore.store( - key = CacheKey( - id = key, - updatedAt = LocalDateTime.now(), - expiresIn = expireIn, - ), - ) + cacheKeysStore.store( + key = CacheKey( + id = key, + updatedAt = LocalDateTime.now(), + expiresIn = expireIn, + ), + ) - block() - } catch (e: Throwable) { - Timber.e(e, "The action related to the cache key has failed: $key") + block() + } catch (e: Throwable) { + Timber.e(e, "The action related to the cache key has failed: $key") - invalidate(key) + invalidate(key) - throw e + throw e + } } } } \ No newline at end of file diff --git a/data/tokens/src/main/kotlin/com/tangem/data/tokens/repository/DefaultNetworksRepository.kt b/data/tokens/src/main/kotlin/com/tangem/data/tokens/repository/DefaultNetworksRepository.kt index fc4296fdb8..44096bdcd7 100644 --- a/data/tokens/src/main/kotlin/com/tangem/data/tokens/repository/DefaultNetworksRepository.kt +++ b/data/tokens/src/main/kotlin/com/tangem/data/tokens/repository/DefaultNetworksRepository.kt @@ -56,8 +56,9 @@ internal class DefaultNetworksRepository( .cancellable() override suspend fun fetchNetworkPendingTransactions(userWalletId: UserWalletId, networks: Set) { + val currencies = getCurrencies(userWalletId, networks) withContext(dispatchers.io) { - fetchNetworksPendingTransactions(userWalletId, networks) + fetchNetworksPendingTransactions(userWalletId, networks, currencies) } } @@ -80,23 +81,28 @@ internal class DefaultNetworksRepository( networks: Set, refresh: Boolean, ) { + val currencies = getCurrencies(userWalletId, networks) coroutineScope { networks .map { network -> async { - fetchNetworkStatusIfCacheExpired(userWalletId, network, refresh) + fetchNetworkStatusIfCacheExpired(userWalletId, network, currencies, refresh) } } .awaitAll() } } - private suspend fun fetchNetworksPendingTransactions(userWalletId: UserWalletId, networks: Set) { + private suspend fun fetchNetworksPendingTransactions( + userWalletId: UserWalletId, + networks: Set, + currencies: Sequence, + ) { coroutineScope { networks .map { network -> async { - fetchNetworkPendingTransactions(userWalletId, network) + fetchNetworkPendingTransactions(userWalletId, network, currencies) } } .awaitAll() @@ -106,22 +112,28 @@ internal class DefaultNetworksRepository( private suspend fun fetchNetworkStatusIfCacheExpired( userWalletId: UserWalletId, network: Network, + currencies: Sequence, refresh: Boolean, ) { cacheRegistry.invokeOnExpire( key = getNetworksStatusesCacheKey(userWalletId, network), skipCache = refresh, - block = { fetchNetworkStatus(userWalletId, network) }, + block = { fetchNetworkStatus(userWalletId, network, currencies) }, ) } - private suspend fun fetchNetworkStatus(userWalletId: UserWalletId, network: Network) { - val currencies = getCurrencies(userWalletId, network) - + private suspend fun fetchNetworkStatus( + userWalletId: UserWalletId, + network: Network, + currencies: Sequence, + ) { val result = walletManagersFacade.update( userWalletId = userWalletId, network = network, - extraTokens = currencies.filterIsInstance().toSet(), + extraTokens = currencies + .filterIsInstance() + .filter { it.network == network } + .toSet(), ) withContext(NonCancellable) { @@ -137,9 +149,11 @@ internal class DefaultNetworksRepository( networksStatusesStore.store(userWalletId, networkStatus) } - private suspend fun fetchNetworkPendingTransactions(userWalletId: UserWalletId, network: Network) { - val currencies = getCurrencies(userWalletId, network) - + private suspend fun fetchNetworkPendingTransactions( + userWalletId: UserWalletId, + network: Network, + currencies: Sequence, + ) { val result = walletManagersFacade.updatePendingTransactions( userWalletId = userWalletId, network = network, @@ -158,7 +172,7 @@ internal class DefaultNetworksRepository( networksStatusesStore.store(userWalletId, networkStatus) } - private suspend fun getCurrencies(userWalletId: UserWalletId, network: Network): Sequence { + private suspend fun getCurrencies(userWalletId: UserWalletId, networks: Set): Sequence { val userWallet = requireNotNull(userWalletsStore.getSyncOrNull(userWalletId)) { "Unable to find user wallet with provided ID: $userWalletId" } @@ -180,7 +194,7 @@ internal class DefaultNetworksRepository( } } - return currencies.filter { it.network == network } + return currencies.filter { networks.contains(it.network) } } private suspend fun invalidateCacheKeyIfNeeded( diff --git a/data/tokens/src/main/kotlin/com/tangem/data/tokens/repository/DefaultQuotesRepository.kt b/data/tokens/src/main/kotlin/com/tangem/data/tokens/repository/DefaultQuotesRepository.kt index bd834c5450..d362189832 100644 --- a/data/tokens/src/main/kotlin/com/tangem/data/tokens/repository/DefaultQuotesRepository.kt +++ b/data/tokens/src/main/kotlin/com/tangem/data/tokens/repository/DefaultQuotesRepository.kt @@ -17,6 +17,8 @@ import com.tangem.domain.tokens.repository.QuotesRepository import com.tangem.utils.coroutines.CoroutineDispatcherProvider import kotlinx.coroutines.ExperimentalCoroutinesApi import kotlinx.coroutines.flow.* +import kotlinx.coroutines.sync.Mutex +import kotlinx.coroutines.sync.withLock import kotlinx.coroutines.withContext internal class DefaultQuotesRepository( @@ -32,6 +34,7 @@ internal class DefaultQuotesRepository( @Volatile private var quotesFetchedForAppCurrency: String? = null + private val mutex = Mutex() @OptIn(ExperimentalCoroutinesApi::class) override fun getQuotesUpdates(currenciesIds: Set): Flow> { @@ -42,7 +45,6 @@ internal class DefaultQuotesRepository( .filterNotNull() .flatMapLatest { appCurrency -> fetchExpiredQuotes(currenciesIds, appCurrency.id, refresh = false) - quotesStore.get(currenciesIds).map(quotesConverter::convertSet) } .cancellable() @@ -79,15 +81,19 @@ internal class DefaultQuotesRepository( appCurrencyId: String, refresh: Boolean, ) { - val expiredCurrenciesIds = filterExpiredCurrenciesIds( - currenciesIds = currenciesIds, - refresh = refresh || quotesFetchedForAppCurrency != appCurrencyId, - ) - if (expiredCurrenciesIds.isEmpty()) return + // TODO("[REDACTED_JIRA]") need refactor working with quotesFetchedForAppCurrency, + // it changes after filterExpiredCurrenciesIds + // calls with different coroutines and lead to fetchQuotes + mutex.withLock { + val expiredCurrenciesIds = filterExpiredCurrenciesIds( + currenciesIds = currenciesIds, + refresh = refresh || quotesFetchedForAppCurrency != appCurrencyId, + ) + if (expiredCurrenciesIds.isEmpty()) return - quotesFetchedForAppCurrency = appCurrencyId - - fetchQuotes(expiredCurrenciesIds, appCurrencyId) + quotesFetchedForAppCurrency = appCurrencyId + fetchQuotes(expiredCurrenciesIds, appCurrencyId) + } } private suspend fun fetchQuotes(rawCurrenciesIds: Set, appCurrencyId: String) { @@ -118,7 +124,6 @@ internal class DefaultQuotesRepository( ): Set { return currenciesIds.fold(hashSetOf()) { acc, currencyId -> val rawCurrencyId = currencyId.rawCurrencyId - if (rawCurrencyId != null && rawCurrencyId !in acc) { cacheRegistry.invokeOnExpire( key = getQuoteCacheKey(rawCurrencyId),