feat(av-cliper): concurrency download for hls-loader

packages/av-cliper/src/data-loader/__tests__/hls-loader.test.ts

@@ -41,3 +41,20 @@ test('hls loader default time', async () => {
   expect(video?.timestamp).toBe(10e6);
   video?.close();
 });
+
+test('hls loader async load m4s files', async () => {
+  const loader = await createHLSLoader(m3u8Url, 5);
+  const [{ actualStartTime, actualEndTime, stream }] = loader.load() ?? [];
+  expect(stream).toBeInstanceOf(ReadableStream);
+  expect([actualStartTime, Math.round(actualEndTime / 1e6)]).toEqual([0, 60]);
+
+  const clip = new MP4Clip(stream);
+  await clip.ready;
+  expect(Math.round(clip.meta.duration / 1e6)).toBe(
+    Math.round((actualEndTime - actualStartTime) / 1e6),
+  );
+
+  const { video } = await clip.tick(10e6);
+  expect(video?.timestamp).toBe(10e6);
+  video?.close();
+});

packages/av-cliper/src/data-loader/hls-loader.ts

@@ -3,7 +3,7 @@ import { Parser } from 'm3u8-parser';
 /**
  * 创建一个 HLS 资源加载器
  */
-export async function createHLSLoader(m3u8URL: string) {
+export async function createHLSLoader(m3u8URL: string, concurrency = 10) {
   const parser = new Parser();
   parser.push(await (await fetch(m3u8URL)).text());
   parser.end();
@@ -17,12 +17,64 @@ export async function createHLSLoader(m3u8URL: string) {
   );
   const base = new URL(m3u8URL, location.href);
 
+  const segmentBufferFetchqueue = {} as Record<string, Promise<ArrayBuffer>>;
+
+  async function downloadSegGroup() {
+    function createTaskQueue(concurrency: number) {
+      let running = 0;
+      const queue = [] as Function[];
+
+      async function runTask(task: Function) {
+        queue.push(task);
+        next();
+      }
+
+      async function next() {
+        if (running < concurrency && queue.length) {
+          const task = queue.shift();
+          running++;
+          try {
+            if (task) {
+              await task();
+            }
+          } catch (err) {
+            console.error(err);
+          }
+          running--;
+          next();
+        }
+      }
+
+      return runTask;
+    }
+
+    async function fetchSegmentBufferPromise(url: string) {
+      return (await fetch(url)).arrayBuffer();
+    }
+
+    const runTask = createTaskQueue(concurrency);
+
+    for (const [, gData] of Object.entries(segGroup)) {
+      for (const [, item] of gData.entries()) {
+        const url = new URL(item.uri, base).href;
+        runTask(
+          () => (segmentBufferFetchqueue[url] = fetchSegmentBufferPromise(url)),
+        );
+      }
+    }
+  }
+
+  async function getSegmentBuffer(url: string) {
+    return segmentBufferFetchqueue[url].then((value) => value);
+  }
+
   return {
     /**
      * 下载期望时间区间的分配数据，封装成流
      * 每个分片包含一个时间段，实际下载的分片数据时长会略大于期望的时间区间
      */
     load(expectStartTime = 0, expectEndTime = Infinity) {
+      downloadSegGroup();
       const filterSegGroup = {} as Record<
         string,
         {
@@ -86,13 +138,8 @@ export async function createHLSLoader(m3u8URL: string) {
                 );
               },
               pull: async (ctrl) => {
-                ctrl.enqueue(
-                  new Uint8Array(
-                    await (
-                      await fetch(new URL(segments[segIdx].uri, base).href)
-                    ).arrayBuffer(),
-                  ),
-                );
+                const url = new URL(segments[segIdx].uri, base).href;
+                ctrl.enqueue(new Uint8Array(await getSegmentBuffer(url)));
                 segIdx += 1;
                 if (segIdx >= segments.length) ctrl.close();
               },