feat: replace task by plain pqueue system with progress

src/core/assets.ts

@@ -4,7 +4,10 @@ import { AssetIndex, PoolFile, Version } from "../utils/types";
 import fs from "fs/promises";
 import { downloadFile, DownloadPool } from "../utils/fetch";
 
-export async function AssetsDownloader(destination: string, versionManifest: Version) {
+export async function AssetsDownloader(
+  destination: string,
+  versionManifest: Version,
+) {
   await ensureDir(path.join(destination, "indexes"));
   await ensureDir(path.join(destination, "objects"));
 
@@ -37,11 +40,11 @@ export async function AssetsDownloader(destination: string, versionManifest: Ver
 
       await ensureDir(path.dirname(assetPath));
 
-      files.push({ url: assetURL, path: assetPath });
+      files.push({ url: assetURL, path: assetPath, size: asset.size });
     }
   }
 
-  const dPool = new DownloadPool(files, 5);
+  const pool = new DownloadPool(files, { concurrency: 5 });
 
-  return dPool;
-}
+  return pool;
+}

src/core/java.ts

@@ -1,8 +1,5 @@
 import path from "path";
-import {
-  DownloadPoolWithCleanup,
-  fetchJson,
-} from "../utils/fetch";
+import { DownloadPool, fetchJson } from "../utils/fetch";
 import fs from "fs/promises";
 import {
   FilesList,
@@ -41,6 +38,7 @@ export async function JavaDownloader(
         files.push({
           url: element.downloads.raw.url,
           path: path.join(destination, key),
+          size: element.downloads.raw.size,
         });
       } else if (element.type === "directory") {
         await fs.mkdir(path.join(destination, key));
@@ -48,12 +46,11 @@ export async function JavaDownloader(
     }
   }
 
-  const dPool = new DownloadPoolWithCleanup(files, 5, async () => {
+  const pool = new DownloadPool(files, { concurrency: 5 }, async () => {
     if (process.platform !== "win32") {
       await fs.chmod(path.join(destination, "bin/java"), 0o777);
-      console.log("chmoded")
     }
   });
 
-  return dPool;
+  return pool;
 }

src/core/libraries.ts

@@ -13,6 +13,7 @@ export function getLibraries(versionManifest: Version, librariesRoot: string) {
         libs.push({
           url: library.downloads.artifact.url,
           path: path.join(librariesRoot, library.downloads.artifact.path),
+          size: library.downloads.artifact.size,
         });
       }
     }
@@ -27,7 +28,7 @@ export async function LibrariesDownloader(
 ) {
   const libs = getLibraries(versionManifest, librariesRoot);
 
-  const dPool = new DownloadPool(libs, 5);
+  const dPool = new DownloadPool(libs, { concurrency: 5 });
 
   return dPool;
 }

src/core/natives.ts

@@ -26,7 +26,7 @@ async function getNatives(versionManifest: Version, tempNativesPath: string) {
           path.basename(native.path),
         );
 
-        files.push({ url: native.url, path: destination });
+        files.push({ url: native.url, path: destination, size: native.size });
       }
     }
   }
@@ -43,12 +43,9 @@ export async function NativesDownloader(
 
   const natives = await getNatives(versionManifest, tempNativesPath);
 
-  const pool = new DownloadAndUnzipPool(
-    natives,
-    5,
-    nativesPath,
-    tempNativesPath,
-  );
+  const pool = new DownloadAndUnzipPool(natives, nativesPath, tempNativesPath, {
+    concurrency: 5,
+  });
 
   return pool;
 }

src/utils/fetch.ts

@@ -1,9 +1,9 @@
 import fs from "fs";
 import { pipeline } from "stream/promises";
 import { PoolFile } from "./types";
-import { Task } from "./task";
 import { ensureDir } from "./fs";
 import path from "path";
+import PQueue, { Options } from "p-queue";
 
 export async function fetchJson<T>(url: string): Promise<T> {
   return await (await fetch(url)).json();
@@ -24,94 +24,40 @@ export async function downloadFile(file: PoolFile) {
     return;
   }
 }
+export class DownloadPool extends PQueue {
+  done = 0;
+  total = 0;
+  doneSize = 0;
+  totalSize = 0;
+  elements: PoolFile[];
+  cleanup: () => Promise<void>;
 
-// export class DownloadPool {
-//   files: PoolFiles;
-//   concurrency: number;
-//   controller: AbortController;
-//   queue?: pQueue;
-
-//   done: number;
-//   total: number;
-//   progressCallback: (done: number, total: number) => void;
-
-//   constructor(
-//     files: PoolFiles,
-//     concurrentDownloads = 5,
-//   ) {
-//     this.files = files;
-//     this.concurrency = concurrentDownloads;
-//     this.controller = new AbortController();
-//     this.queue = new pQueue({ concurrency: this.concurrency });
-//     this.done = 0;
-//     this.total = files.length;
-//   }
-
-//   async run() {
-//     try {
-//       for (const file of this.files) {
-//         this.queue.add(async () => {
-//           await downloadFile(file, this.controller.signal);
-//         });
-//       }
-//     } catch (error) {
-//       if (error.name !== "AbortError") throw error;
-//       return;
-//     }
-
-//     this.queue.on("completed", () => {
-//       this.done++;
-//       this.progressCallback(this.done, this.total);
-//     });
-
-//     await this.queue.onIdle();
-//   }
-
-//   cancel() {
-//     this.queue.clear();
-//     this.controller.abort();
-//   }
-// }
-
-export class DownloadPool extends Task<PoolFile> {
-  files: PoolFile[];
-
-  constructor(files: PoolFile[], concurrency: number) {
-    super(concurrency);
-    this.files = files;
-    this.total = files.length;
-  }
-
-  async run() {
-    await this._run(downloadFile, this.files);
-  }
-}
-
-// export class AssetsDownloader extends DownloadPool {
-//   assets: PoolFile[];
-//   constructor(assets: PoolFile[], concurrency: number) {
-//     super(assets, concurrency);
-//     this.assets = assets;
-//   }
-
-//   async run() {
-//     await this._run(downloadAssets, this.assets);
-//   }
-// }
-
-export class DownloadPoolWithCleanup extends DownloadPool {
-  onCleanup: () => Promise<void>;
   constructor(
-    files: PoolFile[],
-    concurrency: number,
-    onCleanup: () => Promise<void>,
+    elements: PoolFile[],
+    options?: Options<null, null>,
+    cleanup?: () => Promise<void>,
   ) {
-    super(files, concurrency);
-    this.onCleanup = onCleanup;
+    super(options);
+    this.elements = elements;
+    this.total = this.elements.length;
+    this.cleanup = cleanup;
   }
 
   async run() {
-    await this._run(downloadFile, this.files);
-    await this.onCleanup();
+    for (const element of this.elements) {
+      this.totalSize += element.size;
+
+      this.add(async () => {
+        await downloadFile(element);
+        // update status here to ensure that it is run before "completed" events listeners
+        this.done++;
+        this.doneSize += element.size;
+      });
+    }
+
+    await this.onIdle();
+    if (this.cleanup) {
+      await this.cleanup();
+    }
   }
-}
+}

src/utils/task.ts

@@ -1,41 +0,0 @@
-import pQueue from "p-queue";
-
-export class Task<T> {
-  controller: AbortController;
-  concurrency: number;
-  queue?: pQueue;
-
-  done: number;
-  total: number;
-  progressCallback: (done: number, total: number) => void;
-
-  constructor(concurrency: number) {
-    this.concurrency = concurrency;
-    this.queue = new pQueue({ concurrency: this.concurrency });
-    this.done = 0;
-    this.total = 0;
-  }
-
-  async _run(f: (element: unknown) => Promise<void>, data: T[]) {
-    for (const element of data) {
-      this.queue.add(() => f(element));
-    }
-
-    this.queue.on("completed", () => {
-      this.done++;
-      this.progressCallback(this.done, this.total);
-    });
-
-    await this.queue.onIdle();
-  }
-
-  cancel() {
-    this.queue.clear();
-  }
-}
-
-// const t = new Task<{ url: string; path: string }>(5);
-
-// class DownloadPool<T> extends Task<T> {}
-
-// const dp = new DownloadPool<{ url: string; path: string }>(5);

src/utils/types.ts

@@ -59,7 +59,7 @@ export type FilesList = {
   };
 };
 
-export type PoolFile = { url: string; path: string };
+export type PoolFile = { url: string; path: string; size?: number };
 
 export type Versions = {
   latest: { release: string; snapshot: string };

src/utils/unzip.ts

@@ -1,28 +1,25 @@
 import AdmZip from "adm-zip";
 import { PoolFile } from "./types";
-import { Task } from "./task";
-import { downloadFile } from "./fetch";
+import { downloadFile, DownloadPool } from "./fetch";
 import path from "path";
+import { Options } from "p-queue";
 
 export function unzipAll(from: string, to: string) {
   const zip = new AdmZip(from);
   zip.extractAllTo(to, true);
 }
 
-export class DownloadAndUnzipPool extends Task<PoolFile> {
-  files: PoolFile[];
-  tempPath: string;
+export class DownloadAndUnzipPool extends DownloadPool {
   destination: string;
+  tempPath: string;
 
   constructor(
-    files: PoolFile[],
-    concurrency: number,
+    elements: PoolFile[],
     destination: string,
     tempPath: string,
+    options?: Options<null, null>,
   ) {
-    super(concurrency);
-    this.files = files;
-    this.total = files.length;
+    super(elements, options);
     this.tempPath = tempPath;
     this.destination = destination;
   }
@@ -37,8 +34,17 @@ export class DownloadAndUnzipPool extends Task<PoolFile> {
   }
 
   async run() {
-    await this._run(async (file: PoolFile) => {
-      await this.downloadAndUnzip(file);
-    }, this.files);
+    for (const element of this.elements) {
+      this.totalSize += element.size;
+
+      this.add(async () => {
+        await this.downloadAndUnzip(element);
+        // update status here to ensure that it is run before "completed" events listeners
+        this.done++;
+        this.doneSize += element.size;
+      });
+    }
+
+    await this.onIdle();
   }
 }

tests/arguments.test.ts

@@ -25,7 +25,7 @@ function offline() {
 // mock os to linux for testing
 vi.stubGlobal("process", { platform: "linux" });
 
-test("parse arguments correctly for 1.21.8", { timeout: 10000 }, async () => {
+test("parse arguments correctly for 1.21.8", { timeout: 0 }, async () => {
   const args = await nlk.core.arguments.generateLaunchArguments(
     await nlk.core.version.getVersionManifest("1.21.8"),
     path.join(import.meta.dirname, "temp/java"),

tests/assets.test.ts

@@ -21,7 +21,7 @@ vi.stubGlobal("process", { platform: "linux" });
 
 test(
   "download assets for 1.21.8 correctly",
-  { timeout: 60 * 1000 },
+  { timeout: 0 },
   async () => {
     const versionManifest = await nlk.core.version.getVersionManifest("1.21.8");
     const assetsDownloader = await nlk.core.AssetsDownloader(assetsPath, versionManifest);

tests/java.test.ts

@@ -21,14 +21,18 @@ vi.stubGlobal("process", { platform: "linux" });
 
 test(
   "install java-runtime-beta for linux correctly",
-  { timeout: 10000 },
+  { timeout: 0 },
   async () => {
     const javaDownloader = await nlk.core.java.JavaDownloader(
       "linux",
       "java-runtime-delta",
       javaPath,
     );
-    javaDownloader.progressCallback = console.log;
+    javaDownloader.on("completed", () => {
+      console.log(
+        `${javaDownloader.done}/${javaDownloader.total} | ${javaDownloader.doneSize}/${javaDownloader.totalSize}`,
+      );
+    });
     await javaDownloader.run();
 
     expect(

tests/libraries.test.ts

@@ -21,12 +21,16 @@ vi.stubGlobal("process", { platform: "linux" });
 
 test(
   "download libraries for 1.21.8 correctly",
-  { timeout: 10000 },
+  { timeout: 0 },
   async () => {
     const versionManifest = await nlk.core.version.getVersionManifest("1.21.8");
-    const LibrariesDownloader = await nlk.core.LibrariesDownloader(librariesPath, versionManifest);
-    LibrariesDownloader.progressCallback = console.log
-    await LibrariesDownloader.run()
+    const librariesDownloader = await nlk.core.LibrariesDownloader(librariesPath, versionManifest);
+    librariesDownloader.on("completed", () => {
+      console.log(
+        `${librariesDownloader.done}/${librariesDownloader.total} | ${librariesDownloader.doneSize}/${librariesDownloader.totalSize}`,
+      );
+    });
+    await librariesDownloader.run()
 
     expect(
       await exists(path.join(librariesPath, "com/fasterxml/jackson/core/jackson-core/2.13.4/jackson-core-2.13.4.jar")),

tests/log4j.test.ts

@@ -17,7 +17,7 @@ vi.stubGlobal("process", { platform: "linux" });
 
 test(
   "download xml file and return correct arguments",
-  { timeout: 10000 },
+  { timeout: 0 },
   async () => {
     const versionManifest = await nlk.core.version.getVersionManifest("1.21.8");
     const arg = await nlk.core.log4j.getArgument(versionManifest, root);

tests/natives.test.ts

@@ -20,13 +20,17 @@ async function exists(path: string) {
 vi.stubGlobal("process", { platform: "linux" });
 
 // old version because newer versions doesn't have natives
-test("download natives for 1.15 correctly", { timeout: 10000 }, async () => {
+test("download natives for 1.15 correctly", { timeout: 0 }, async () => {
   const versionManifest = await nlk.core.version.getVersionManifest("1.15");
   const nativesDownloader = await nlk.core.NativesDownloader(
     nativesPath,
     versionManifest,
   );
-  nativesDownloader.progressCallback = console.log;
+  nativesDownloader.on("completed", () => {
+    console.log(
+      `${nativesDownloader.done}/${nativesDownloader.total} | ${nativesDownloader.doneSize}/${nativesDownloader.totalSize}`,
+    );
+  });
   await nativesDownloader.run();
 
   expect(

tests/versionJar.test.ts

@@ -19,7 +19,7 @@ async function exists(path: string) {
 // mock os to linux for testing
 vi.stubGlobal("process", { platform: "linux" });
 
-test("parse arguments correctly for 1.21.8", { timeout: 10000 }, async () => {
+test("parse arguments correctly for 1.21.8", { timeout: 0 }, async () => {
   const versionManifest = await nlk.core.version.getVersionManifest("1.21.8");
   await nlk.core.version.downloadJar(
     versionManifest,