refactor: extract structs for downloading tarballs and npm registry packuments (#24067)

1 files changed, 16 insertions, 12 deletions

diff --git a/cli/npm/managed/resolvers/common.rs b/cli/npm/managed/resolvers/common.rs
index b010bdd7c..2d540accd 100644
--- a/cli/npm/managed/resolvers/common.rs
+++ b/cli/npm/managed/resolvers/common.rs
@@ -12,7 +12,7 @@ use deno_ast::ModuleSpecifier;
 use deno_core::anyhow::Context;
 use deno_core::error::AnyError;
 use deno_core::futures;
-use deno_core::unsync::spawn;
+use deno_core::futures::StreamExt;
 use deno_core::url::Url;
 use deno_npm::NpmPackageCacheFolderId;
 use deno_npm::NpmPackageId;
@@ -21,7 +21,8 @@ use deno_runtime::deno_fs::FileSystem;
 use deno_runtime::deno_node::NodePermissions;
 use deno_runtime::deno_node::NodeResolutionMode;
 
-use super::super::cache::NpmCache;
+use crate::http_util::HttpClient;
+use crate::npm::managed::cache::TarballCache;
 
 /// Part of the resolution that interacts with the file system.
 #[async_trait]
@@ -49,7 +50,10 @@ pub trait NpmPackageFsResolver: Send + Sync {
     specifier: &ModuleSpecifier,
   ) -> Result<Option<NpmPackageCacheFolderId>, AnyError>;
 
-  async fn cache_packages(&self) -> Result<(), AnyError>;
+  async fn cache_packages(
+    &self,
+    http_client: &Arc<HttpClient>,
+  ) -> Result<(), AnyError>;
 
   fn ensure_read_permission(
     &self,
@@ -126,20 +130,20 @@ impl RegistryReadPermissionChecker {
 /// Caches all the packages in parallel.
 pub async fn cache_packages(
   packages: Vec<NpmResolutionPackage>,
-  cache: &Arc<NpmCache>,
+  tarball_cache: &Arc<TarballCache>,
+  http_client: &Arc<HttpClient>,
 ) -> Result<(), AnyError> {
-  let mut handles = Vec::with_capacity(packages.len());
+  let mut futures_unordered = futures::stream::FuturesUnordered::new();
   for package in packages {
-    let cache = cache.clone();
-    let handle = spawn(async move {
-      cache.ensure_package(&package.id.nv, &package.dist).await
+    futures_unordered.push(async move {
+      tarball_cache
+        .ensure_package(&package.id.nv, &package.dist, http_client)
+        .await
     });
-    handles.push(handle);
   }
-  let results = futures::future::join_all(handles).await;
-  for result in results {
+  while let Some(result) = futures_unordered.next().await {
     // surface the first error
-    result??;
+    result?;
   }
   Ok(())
 }