diff --git a/library/src/commonMain/kotlin/com/lagradost/cloudstream3/ParCollections.kt b/library/src/commonMain/kotlin/com/lagradost/cloudstream3/ParCollections.kt index 9e6cb99e5be..58900cf3e68 100644 --- a/library/src/commonMain/kotlin/com/lagradost/cloudstream3/ParCollections.kt +++ b/library/src/commonMain/kotlin/com/lagradost/cloudstream3/ParCollections.kt @@ -1,19 +1,35 @@ package com.lagradost.cloudstream3 import com.lagradost.cloudstream3.mvvm.logError -import kotlinx.coroutines.* -import kotlin.coroutines.cancellation.CancellationException +import kotlinx.coroutines.CancellationException +import kotlinx.coroutines.async +import kotlinx.coroutines.awaitAll +import kotlinx.coroutines.coroutineScope +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.sync.Semaphore +import kotlinx.coroutines.sync.withPermit + +private const val DEFAULT_CONCURRENCY = 4 + +@Deprecated("Use amap with concurrency parameter", level = DeprecationLevel.HIDDEN) +@Throws(CancellationException::class) +suspend fun Map.amap(f: suspend (Map.Entry) -> R): List = + amap(DEFAULT_CONCURRENCY, f) /** * Short for "Asynchronous Map", runs on all values concurrently, * this means that if you are not doing networking, you should use a regular map */ @Throws(CancellationException::class) -suspend fun Map.amap(f: suspend (Map.Entry) -> R): List = - coroutineScope { - ensureActive() - map { async { f(it) } }.map { it.await() } - } +suspend fun Map.amap( + concurrency: Int = DEFAULT_CONCURRENCY, + f: suspend (Map.Entry) -> R, +): List = coroutineScope { + val semaphore = Semaphore(concurrency) + map { entry -> + async { semaphore.withPermit { f(entry) } } + }.awaitAll() +} /** * Short for "Asynchronous Parallel Map", but is not really parallel, only concurrent. @@ -28,16 +44,25 @@ fun Map.apmap(f: suspend (Map.Entry) -> R): List = map { async { f(it) } }.map { it.await() } } +@Deprecated("Use amap with concurrency parameter", level = DeprecationLevel.HIDDEN) +@Throws(CancellationException::class) +suspend fun List.amap(f: suspend (A) -> B): List = + amap(DEFAULT_CONCURRENCY, f) + /** * Short for "Asynchronous Map", runs on all values concurrently, * this means that if you are not doing networking, you should use a regular map */ @Throws(CancellationException::class) -suspend fun List.amap(f: suspend (A) -> B): List = - coroutineScope { - ensureActive() - map { async { f(it) } }.map { it.await() } - } +suspend fun List.amap( + concurrency: Int = DEFAULT_CONCURRENCY, + f: suspend (A) -> B, +): List = coroutineScope { + val semaphore = Semaphore(concurrency) + map { item -> + async { semaphore.withPermit { f(item) } } + }.awaitAll() +} /** * Short for "Asynchronous Parallel Map", but is not really parallel, only concurrent. @@ -65,19 +90,28 @@ fun List.apmapIndexed(f: suspend (index: Int, A) -> B): List = runB mapIndexed { index, a -> async { f(index, a) } }.map { it.await() } } +@Deprecated("Use amapIndexed with concurrency parameter", level = DeprecationLevel.HIDDEN) +@Throws(CancellationException::class) +suspend fun List.amapIndexed(f: suspend (index: Int, A) -> B): List = + amapIndexed(DEFAULT_CONCURRENCY, f) + /** * Short for "Asynchronous Map" with an Index, runs on all values concurrently, * this means that if you are not doing networking, you should use a regular mapIndexed */ @Throws(CancellationException::class) -suspend fun List.amapIndexed(f: suspend (index: Int, A) -> B): List = - coroutineScope { - ensureActive() - mapIndexed { index, a -> async { f(index, a) } }.map { it.await() } - } +suspend fun List.amapIndexed( + concurrency: Int = DEFAULT_CONCURRENCY, + f: suspend (index: Int, A) -> B, +): List = coroutineScope { + val semaphore = Semaphore(concurrency) + mapIndexed { index, item -> + async { semaphore.withPermit { f(index, item) } } + }.awaitAll() +} /** - * Short for "Argument Asynchronous Map" because it allows for a variadic number of paramaters. + * Short for "Argument Asynchronous Map" because it allows for a variadic number of parameters. * * Runs all different functions at the same time and awaits for all to be finished, then returns * a list of all those items or null if they fail. However Unit is often used. @@ -103,6 +137,12 @@ fun argamap( }.map { it.await() } } +@Deprecated("Use runAllAsync with concurrency parameter", level = DeprecationLevel.HIDDEN) +@Throws(CancellationException::class) +suspend fun runAllAsync( + vararg transforms: suspend () -> R, +): List = runAllAsync(*transforms, concurrency = DEFAULT_CONCURRENCY) + /** * Runs all different functions at the same time and awaits for all to be finished, then returns * a list of all those items or null if they fail. However Unit is often used. @@ -110,16 +150,21 @@ fun argamap( @Throws(CancellationException::class) suspend fun runAllAsync( vararg transforms: suspend () -> R, -) : List = coroutineScope { - ensureActive() + concurrency: Int = DEFAULT_CONCURRENCY, +): List = coroutineScope { + val semaphore = Semaphore(concurrency) transforms.map { fn -> async { - try { - fn.invoke() - } catch (e: Exception) { - logError(e) - null + semaphore.withPermit { + try { + fn() + } catch (e: CancellationException) { + throw e + } catch (ex: Exception) { + logError(ex) + null + } } } - }.map { it.await() } -} \ No newline at end of file + }.awaitAll() +}