From ade13382ecc1945af5b60289555bdd829dc3a6be Mon Sep 17 00:00:00 2001 From: Luna712 <142361265+Luna712@users.noreply.github.com> Date: Sun, 28 Jun 2026 16:28:29 -0600 Subject: [PATCH 1/3] Improve amap methods --- .../lagradost/cloudstream3/ParCollections.kt | 99 ++++++++++++++----- 1 file changed, 72 insertions(+), 27 deletions(-) diff --git a/library/src/commonMain/kotlin/com/lagradost/cloudstream3/ParCollections.kt b/library/src/commonMain/kotlin/com/lagradost/cloudstream3/ParCollections.kt index 9e6cb99e5be..8e2f32d575d 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,23 +137,34 @@ 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(DEFAULT_CONCURRENCY, *transforms) + /** * 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. */ @Throws(CancellationException::class) suspend fun runAllAsync( + concurrency: Int = DEFAULT_CONCURRENCY, vararg transforms: suspend () -> R, -) : List = coroutineScope { - ensureActive() +): 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() +} From 4ceeddbeaddd507c80d6e1cf65f47edeaf5dd649 Mon Sep 17 00:00:00 2001 From: Luna712 <142361265+Luna712@users.noreply.github.com> Date: Sun, 28 Jun 2026 16:38:30 -0600 Subject: [PATCH 2/3] Test --- .../kotlin/com/lagradost/cloudstream3/ParCollections.kt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/library/src/commonMain/kotlin/com/lagradost/cloudstream3/ParCollections.kt b/library/src/commonMain/kotlin/com/lagradost/cloudstream3/ParCollections.kt index 8e2f32d575d..054f9346497 100644 --- a/library/src/commonMain/kotlin/com/lagradost/cloudstream3/ParCollections.kt +++ b/library/src/commonMain/kotlin/com/lagradost/cloudstream3/ParCollections.kt @@ -137,7 +137,7 @@ fun argamap( }.map { it.await() } } -@Deprecated("Use runAllAsync with concurrency parameter", level = DeprecationLevel.HIDDEN) +// @Deprecated("Use runAllAsync with concurrency parameter", level = DeprecationLevel.HIDDEN) @Throws(CancellationException::class) suspend fun runAllAsync( vararg transforms: suspend () -> R, From e50cf7da373dd0488aa3256b2c69ad5563149743 Mon Sep 17 00:00:00 2001 From: Luna712 <142361265+Luna712@users.noreply.github.com> Date: Sun, 28 Jun 2026 16:42:00 -0600 Subject: [PATCH 3/3] - --- .../kotlin/com/lagradost/cloudstream3/ParCollections.kt | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/library/src/commonMain/kotlin/com/lagradost/cloudstream3/ParCollections.kt b/library/src/commonMain/kotlin/com/lagradost/cloudstream3/ParCollections.kt index 054f9346497..58900cf3e68 100644 --- a/library/src/commonMain/kotlin/com/lagradost/cloudstream3/ParCollections.kt +++ b/library/src/commonMain/kotlin/com/lagradost/cloudstream3/ParCollections.kt @@ -137,11 +137,11 @@ fun argamap( }.map { it.await() } } -// @Deprecated("Use runAllAsync with concurrency parameter", level = DeprecationLevel.HIDDEN) +@Deprecated("Use runAllAsync with concurrency parameter", level = DeprecationLevel.HIDDEN) @Throws(CancellationException::class) suspend fun runAllAsync( vararg transforms: suspend () -> R, -): List = runAllAsync(DEFAULT_CONCURRENCY, *transforms) +): List = runAllAsync(*transforms, concurrency = DEFAULT_CONCURRENCY) /** * Runs all different functions at the same time and awaits for all to be finished, then returns @@ -149,8 +149,8 @@ suspend fun runAllAsync( */ @Throws(CancellationException::class) suspend fun runAllAsync( - concurrency: Int = DEFAULT_CONCURRENCY, vararg transforms: suspend () -> R, + concurrency: Int = DEFAULT_CONCURRENCY, ): List = coroutineScope { val semaphore = Semaphore(concurrency) transforms.map { fn ->