GiftedHacker/tangem-app-android-audited
1
0
Fork 0
Code Issues Pull requests Projects Releases 1 Packages Wiki Activity Actions

Updated on 2026-08-14

Browse source
This commit is contained in:
Tangem 2024-07-16 11:07:58 +03:00
parent f7412dfd86
commit 0cd5575681
20 changed files with 437 additions and 162 deletions
Download patch file Download diff file Expand all files Collapse all files

12
core/pagination/src/main/java/com/tangem/pagination/BatchAction.kt
View file

@ -1,5 +1,7 @@
package com.tangem.pagination
import java.util.UUID
/**
* Action that can be dispatched to [BatchListSource].
*
@ -7,7 +9,7 @@ package com.tangem.pagination
* @param TKey type of the key of the batch.
* @param TUpdate type of the update request.
*/
sealed class BatchAction<TKey, TRequestParams, TUpdate> {
sealed class BatchAction<out TKey, out TRequestParams, out TUpdate> {
/**
* Action to load the first batch.
@ -34,10 +36,18 @@ sealed class BatchAction<TKey, TRequestParams, TUpdate> {
*
* @param keys keys of the batches to update.
* @param updateRequest request to update the batches.
* @param async true if the request doesn't require to synchronize on specific batches in order to fetch update
* data, this request will be delegated to fetchAsync method in [BatchUpdateFetcher],
* false if request requires to hold the current batches data until fetch + update is completed
* @param operationId the unique identifier of the request.
* Only one request with the same hash can be executed at a time,
* the rest of the requests will be canceled as long as there is a request with this hash in progress.
*/
class UpdateBatches<TKey, TUpdate>(
val keys: Set<TKey>,
val updateRequest: TUpdate,
val async: Boolean = false,
val operationId: String = UUID.randomUUID().toString(),
) : BatchAction<TKey, Nothing, TUpdate>()
/**

165
core/pagination/src/main/java/com/tangem/pagination/BatchListSource.kt
View file

@ -75,6 +75,8 @@ private class DefaultBatchListSource<TKey, TData, TRequestParams : Any, TUpdate>
private val scope = context.coroutineScope
private val updateJobs = MutableStateFlow<List<Pair<BatchAction.UpdateBatches<TKey, TUpdate>, Job>>>(emptyList())
private val updateAsyncJobs =
MutableStateFlow<List<Pair<BatchAction.UpdateBatches<TKey, TUpdate>, Job>>>(emptyList())
private val waitingUpdateJobs =
MutableStateFlow<List<Pair<BatchAction.UpdateBatches<TKey, TUpdate>, Job>>>(emptyList())
@ -127,38 +129,13 @@ private class DefaultBatchListSource<TKey, TData, TRequestParams : Any, TUpdate>
}
is BatchAction.UpdateBatches -> {
if (updateFetcher == null) return
// If the request with the same operationId is in progress, skip the request
if (updateInProgressExists(action.operationId)) return
scope.launch(fetchDispatcher) {
// Lazily start a job so we can avoid batch update collisions
// by waiting for other tasks with the same keys to complete
val job = launch(start = CoroutineStart.LAZY) {
updateBatchesTask(action)
}
val actionJob = action to job
waitingUpdateJobs.update { it + actionJob }
// Wait for other update tasks that mutate batches with the same keys
updateJobs.first { workingJobs ->
action.keys.intersect(workingJobs.map { it.first.keys }.flatten().toSet()).isEmpty()
}
waitingUpdateJobs.update { it - actionJob }
// No other task are mutating batches with the same keys, so we can start a job
val started = job.start()
if (started) {
updateJobs.update { it + actionJob }
job.invokeOnCompletion { cause ->
// If the job was cancelled it is up to a canceller to remove job from the updateJobs list
if (cause !is CancellationException) {
updateJobs.update { it - actionJob }
}
}
}
if (action.async) {
collectAsyncUpdateAction(action)
} else {
collectSyncUpdateAction(action)
}
}
BatchAction.CancelAllUpdates -> {
@ -176,6 +153,57 @@ private class DefaultBatchListSource<TKey, TData, TRequestParams : Any, TUpdate>
}
}
private fun collectAsyncUpdateAction(action: BatchAction.UpdateBatches<TKey, TUpdate>) {
val job = scope.launch(fetchDispatcher) {
updateBatchesAsyncTask(action)
}
val actionJob = action to job
updateAsyncJobs.update { it + actionJob }
job.invokeOnCompletion { cause ->
// If the job was cancelled it is up to a canceller to remove job from the updateJobs list
if (cause !is CancellationException) {
updateAsyncJobs.update { it - actionJob }
}
}
}
private fun collectSyncUpdateAction(action: BatchAction.UpdateBatches<TKey, TUpdate>) {
// Lazily start a job so we can avoid batch update collisions
// by waiting for other tasks with the same keys to complete
val job = scope.launch(fetchDispatcher, start = CoroutineStart.LAZY) {
updateBatchesTask(action)
}
val actionJob = action to job
waitingUpdateJobs.update { it + actionJob }
scope.launch(fetchDispatcher) {
// Wait for other update tasks that mutate batches with the same keys
updateJobs.first { workingJobs ->
action.keys.intersect(workingJobs.map { it.first.keys }.flatten().toSet()).isEmpty()
}
waitingUpdateJobs.update { it - actionJob }
// No other task are mutating batches with the same keys, so we can start a job
val started = job.start()
if (started) {
updateJobs.update { it + actionJob }
job.invokeOnCompletion { cause ->
// If the job was cancelled it is up to a canceller to remove job from the updateJobs list
if (cause !is CancellationException) {
updateJobs.update { it - actionJob }
}
}
}
}
}
private suspend fun reloadTask(action: BatchAction.Reload<TRequestParams>) {
state.value = BatchListState(
data = emptyList(),
@ -184,7 +212,10 @@ private class DefaultBatchListSource<TKey, TData, TRequestParams : Any, TUpdate>
val res = runCatching {
batchFetcher.fetchFirst(action.requestParams)
}.getOrElse { BatchFetchResult.Error(it) }
}.getOrElse {
currentCoroutineContext().ensureActive()
BatchFetchResult.Error(it)
}
state.value = when (res) {
is BatchFetchResult.Success -> {
@ -275,6 +306,7 @@ private class DefaultBatchListSource<TKey, TData, TRequestParams : Any, TUpdate>
updateRequest = action.updateRequest,
)
} catch (t: Throwable) {
currentCoroutineContext().ensureActive()
BatchUpdateResult.Error(t)
}
@ -292,7 +324,64 @@ private class DefaultBatchListSource<TKey, TData, TRequestParams : Any, TUpdate>
updateResults.emit(action.updateRequest to result)
}
private suspend fun updateBatchesAsyncTask(action: BatchAction.UpdateBatches<TKey, TUpdate>) {
if (updateFetcher == null) return
val batches = state.value.data
val batchesToUpdate = batches.filter { action.keys.contains(it.key) }
val updateContext = UpdateContext(request = action.updateRequest, action.keys)
with(updateFetcher) {
updateContext.fetchUpdateAsync(batchesToUpdate, action.updateRequest)
}
}
@Suppress("FunctionNaming")
private fun UpdateContext(request: TUpdate, keysToUpdate: Set<TKey>) =
object : BatchUpdateFetcher.UpdateContext<TKey, TData> {
override suspend fun update(update: List<Batch<TKey, TData>>.() -> BatchUpdateResult<TKey, TData>) {
val stateToFetchUpdateBasedOn = state.value.data.filter {
keysToUpdate.contains(it.key)
}
val result = runCatching {
stateToFetchUpdateBasedOn.update()
}.getOrElse {
currentCoroutineContext().ensureActive()
BatchUpdateResult.Error(it)
}
if (result is BatchUpdateResult.Success) {
state.update { currentState ->
val resMap = result.data.associateBy { it.key }
currentState.copy(
data = currentState.data.map {
resMap[it.key] ?: it
},
)
}
}
updateResults.emit(request to result)
}
}
private fun updateInProgressExists(operationId: String): Boolean {
return updateAsyncJobs.value.any { it.first.operationId == operationId } ||
updateJobs.value.any { it.first.operationId == operationId } ||
waitingUpdateJobs.value.any { it.first.operationId == operationId }
}
private fun stopAllUpdates() {
updateAsyncJobs.update { actionAsyncJobs ->
actionAsyncJobs.forEach {
it.second.cancel()
}
emptyList()
}
updateJobs.update { actionJobs ->
waitingUpdateJobs.update { waitingActionJobs ->
waitingActionJobs.forEach {
@ -308,6 +397,18 @@ private class DefaultBatchListSource<TKey, TData, TRequestParams : Any, TUpdate>
}
private fun stopUpdates(predicate: (BatchAction.UpdateBatches<TKey, TUpdate>) -> Boolean) {
updateAsyncJobs.update { actionAsyncJobs ->
actionAsyncJobs.mapNotNull {
if (predicate(it.first)) {
it.second.cancel()
null
} else {
it
}
}
emptyList()
}
updateJobs.update { actionJobs ->
waitingUpdateJobs.update { waitingActionJobs ->
waitingActionJobs.mapNotNull {

44
core/pagination/src/main/java/com/tangem/pagination/BatchUpdateFetcher.kt
View file

@ -8,7 +8,7 @@ package com.tangem.pagination
* @param TData type of the data.
* @param TUpdate type of the update request.
*/
fun interface BatchUpdateFetcher<TKey, TData, TUpdate> {
interface BatchUpdateFetcher<TKey, TData, TUpdate> {
/**
* Fetches updates for a batch of data.
@ -19,5 +19,45 @@ fun interface BatchUpdateFetcher<TKey, TData, TUpdate> {
* @param updateRequest request to update the data.
* @return result of the update operation.
*/
suspend fun fetchUpdate(toUpdate: List<Batch<TKey, TData>>, updateRequest: TUpdate): BatchUpdateResult<TKey, TData>
suspend fun fetchUpdate(
toUpdate: List<Batch<TKey, TData>>,
updateRequest: TUpdate,
): BatchUpdateResult<TKey, TData> = BatchUpdateResult.Error(NotImplementedError())
/**
* Fetches updates for a batch of data asynchronously.
* To update the data, use the [UpdateContext.update] method.
* [UpdateContext.update] could be called as many times as you want.
*
* Note that the result batch key as a result of executing the method must be presented in the [toUpdate] list,
* otherwise, updates will not be performed.
*
* @param toUpdate list of batches to update. **Attention** Data may be outdated and should be used only to make
* a request for an update, not for the actual update operation. For the actual update operation, use the batches
* provided by [UpdateContext.update].
* @param updateRequest request to update the data.
*/
suspend fun UpdateContext<TKey, TData>.fetchUpdateAsync(
toUpdate: List<Batch<TKey, TData>>,
updateRequest: TUpdate,
) {
}
/**
* Context for updating the data.
* Used by [BatchListSource] to provide a way to update batches by [fetchUpdateAsync] method.
*/
interface UpdateContext<TKey, TData> {
/**
* Updates the data of the batch.
* Could be called as many times as you want.
*
* Input batches keys and the data could not always be the same as the keys of the [toUpdate] list in
* [fetchUpdateAsync] method, but the provided set of keys will always be a subset of the [toUpdate] list.
*
* @param update lambda to update the data.
*/
suspend fun update(update: List<Batch<TKey, TData>>.() -> BatchUpdateResult<TKey, TData>)
}
}

35
core/pagination/src/main/java/com/tangem/pagination/fetcher/LimitOffsetBatchFetcher.kt
View file

@ -2,6 +2,8 @@ package com.tangem.pagination.fetcher
import com.tangem.pagination.BatchFetchResult
import com.tangem.pagination.exception.EndOfPaginationException
import kotlinx.coroutines.currentCoroutineContext
import kotlinx.coroutines.ensureActive
import kotlinx.coroutines.flow.MutableStateFlow
/**
@ -12,32 +14,42 @@ import kotlinx.coroutines.flow.MutableStateFlow
*
* @property prefetchDistance number of items to fetch for the first batch.
* @property batchSize size of the batch.
* @property fetch function that fetches the data.
* @property subFetcher function that fetches the data.
*/
class LimitOffsetBatchFetcher<TRequestParams : Any, TData>(
private val prefetchDistance: Int,
private val batchSize: Int,
private val fetch: suspend (request: Request<TRequestParams>) -> BatchFetchResult<TData>,
private val subFetcher: SubFetcher<TRequestParams, TData>,
) : BatchFetcher<TRequestParams, TData> {
data class Request<TRequest>(
data class Request<TRequestParams>(
val limit: Int,
val offset: Int,
val request: TRequest,
val params: TRequestParams,
)
fun interface SubFetcher<TRequestParams : Any, TData> {
suspend fun fetch(
request: Request<TRequestParams>,
lastResult: BatchFetchResult<TData>?,
): BatchFetchResult<TData>
}
private val lastRequest = MutableStateFlow<Request<TRequestParams>?>(null)
override suspend fun fetchFirst(requestParams: TRequestParams): BatchFetchResult<TData> {
val req = Request(
offset = 0,
limit = prefetchDistance,
request = requestParams,
params = requestParams,
)
val res = runCatching {
fetch(req)
}.getOrElse { BatchFetchResult.Error(it) }
subFetcher.fetch(req, null)
}.getOrElse {
currentCoroutineContext().ensureActive()
BatchFetchResult.Error(it)
}
lastRequest.value = req
return res
@ -58,15 +70,18 @@ class LimitOffsetBatchFetcher<TRequestParams : Any, TData>(
Request(
offset = last.offset + last.limit,
limit = batchSize,
request = overrideRequestParams ?: last.request,
params = overrideRequestParams ?: last.params,
)
} else {
last
}
val res = runCatching {
fetch(req)
}.getOrElse { BatchFetchResult.Error(it) }
subFetcher.fetch(req, lastResult)
}.getOrElse {
currentCoroutineContext().ensureActive()
BatchFetchResult.Error(it)
}
lastRequest.value = req
return res