Hls: manually fetch segments and control playback stream level

p2p-media-loader-demo/src/App.tsx

@@ -18,7 +18,7 @@ type Player = (typeof players)[number];
 type ShakaPlayer = shaka.Player;
 type ExtendedWindow = Window & { videoPlayer?: { destroy?: () => void } };
 
-const videoUrl = {
+const streamUrl = {
   bigBunnyBuck: "https://test-streams.mux.dev/x36xhzz/x36xhzz.m3u8",
   byteRangeVideo:
     "https://devstreaming-cdn.apple.com/videos/streaming/examples/bipbop_16x9/bipbop_16x9_variant.m3u8",
@@ -48,7 +48,7 @@ function App() {
   const [playerType, setPlayerType] = useState<Player | undefined>(
     localStorage.player
   );
-  const [url, setUrl] = useState<string>(localStorage.videoUrl);
+  const [url, setUrl] = useState<string>(localStorage.streamUrl);
   const shakaEngine = useRef<ShakaEngine>(new ShakaEngine(shakaLib));
   const hlsEngine = useRef<HlsJsEngine>(new HlsJsEngine());
   const shakaInstance = useRef<shaka.Player>();
@@ -67,6 +67,10 @@ function App() {
       localStorage.player = "hls-dplayer";
       setPlayerType("hls-dplayer");
     }
+    if (!localStorage.streamUrl) {
+      localStorage.streamUrl = streamUrl.live2;
+      setUrl(streamUrl.live2);
+    }
 
     createNewPlayer();
   }, [playerType]);
@@ -171,14 +175,14 @@ function App() {
   };
 
   const onVideoUrlChange = (url: string) => {
-    localStorage.videoUrl = url;
+    localStorage.streamUrl = url;
     setUrl(url);
   };
 
   const createNewPlayer = () => {
     if (!localStorage.videoUrl) {
-      localStorage.videoUrl = videoUrl.live2;
-      setUrl(videoUrl.live2);
+      localStorage.streamUrl = streamUrl.live2;
+      setUrl(streamUrl.live2);
     }
     switch (playerType) {
       case "hls-dplayer":
@@ -232,7 +236,7 @@ function App() {
             value={url}
             onChange={(event) => onVideoUrlChange(event.target.value)}
           >
-            {Object.entries(videoUrl).map(([name, url]) => {
+            {Object.entries(streamUrl).map(([name, url]) => {
               return (
                 <option key={name} value={url}>
                   {name}

packages/p2p-media-loader-hlsjs/src/services/engine.ts

@@ -23,10 +23,17 @@ export class Engine {
       this.segmentManager.processMasterManifest(data);
     });
 
-    hls.on("hlsLevelLoaded" as Events.LEVEL_LOADED, (event, data) => {
-      this.segmentManager.setPlaylist(data);
+    hls.on("hlsLevelUpdated" as Events.LEVEL_UPDATED, (event, data) => {
+      this.segmentManager.updatePlaylist(data);
     });
 
+    hls.on(
+      "hlsAudioTrackLoaded" as Events.AUDIO_TRACK_LOADED,
+      (event, data) => {
+        this.segmentManager.updatePlaylist(data);
+      }
+    );
+
     hls.on("hlsDestroying" as Events.DESTROYING, () => {
       this.debugDestroying("Hls destroying");
       this.destroy();

packages/p2p-media-loader-hlsjs/src/services/fragment-loader.ts

@@ -1,29 +1,51 @@
 import type {
+  FragmentLoaderContext,
+  HlsConfig,
   Loader,
   LoaderCallbacks,
   LoaderConfiguration,
-  LoaderStats,
-  FragmentLoaderContext,
-  HlsConfig,
   LoaderContext,
+  LoaderStats,
 } from "hls.js";
 import { SegmentManager } from "./segment-mananger";
-import { Segment } from "./playlist";
+import { ByteRange, Segment } from "./playlist";
 import Debug from "debug";
 
+const DEFAULT_DOWNLOAD_LATENCY = 10;
+
 export class FragmentLoaderBase implements Loader<FragmentLoaderContext> {
   context!: FragmentLoaderContext;
   config!: LoaderConfiguration | null;
   callbacks!: LoaderCallbacks<FragmentLoaderContext> | null;
   stats: LoaderStats;
-  defaultLoader: Loader<LoaderContext>;
+  createDefaultLoader: () => Loader<LoaderContext>;
+  defaultLoader?: Loader<LoaderContext>;
   segmentManager: SegmentManager;
-  private debug = Debug("p2pml:fragment-loader");
+  response?: {
+    status: number;
+    ok: boolean;
+    url: string;
+    data: ArrayBuffer;
+  };
+  abortController: AbortController = new AbortController();
+  private debug = Debug("hls:fragment-loading");
 
   constructor(config: HlsConfig, segmentManager: SegmentManager) {
     this.segmentManager = segmentManager;
-    this.defaultLoader = new config.loader(config);
-    this.stats = this.defaultLoader.stats;
+    this.createDefaultLoader = () => new config.loader(config);
+    this.stats = {
+      aborted: false,
+      chunkCount: 0,
+      loading: { start: 0, first: 0, end: 0 },
+      buffering: { start: 0, first: 0, end: 0 },
+      parsing: { start: 0, end: 0 },
+      // set total and loaded to 1 to prevent hls.js
+      // on progress loading monitoring in AbrController
+      total: 1,
+      loaded: 1,
+      bwEstimate: 0,
+      retry: 0,
+    };
   }
 
   async load(
@@ -34,32 +56,153 @@ export class FragmentLoaderBase implements Loader<FragmentLoaderContext> {
     this.context = context;
     this.config = config;
     this.callbacks = callbacks;
-    this.defaultLoader.load(context, config, {
-      ...callbacks,
-      onSuccess: (response, stats, context, networkDetails) => {
-        const { rangeStart: start, rangeEnd: end } = context;
-        const segmentId = Segment.getSegmentLocalId(context.url, {
-          start,
-          end: end !== undefined ? end - 1 : undefined,
-        });
-        const playlist = this.segmentManager.getPlaylistBySegmentId(segmentId);
-        this.debug(
-          "downloaded segment from playlist\n",
-          `playlist v: ${playlist?.index}\n`,
-          `segment: `,
-          playlist?.segments.get(segmentId)?.index
-        );
-
-        return callbacks.onSuccess(response, stats, context, networkDetails);
-      },
+    const stats = this.stats;
+
+    const playlist = this.identifyPlaylist(context);
+    if (!playlist) {
+      this.defaultLoader = this.createDefaultLoader();
+      this.defaultLoader.stats = this.stats;
+      this.defaultLoader?.load(context, config, callbacks);
+      return;
+    }
+
+    try {
+      const byteRange = getByteRange(context.rangeStart, context.rangeEnd);
+      this.response = await this.fetchSegment(context.url, byteRange);
+    } catch (error) {
+      if (this.stats.aborted) return;
+      return this.handleError(error);
+    }
+    if (!this.response) return;
+    const loadedBytes = this.response.data.byteLength;
+
+    stats.loading = getLoadingStat({
+      targetBitrate: 4947980 * (10 / 6.8),
+      loadingEndTime: performance.now(),
+      loadedBytes,
+    });
+    stats.total = stats.loaded = loadedBytes;
+
+    callbacks.onSuccess(this.response, this.stats, context, this.response);
+  }
+
+  private identifyPlaylist(context: LoaderContext) {
+    const { rangeStart: start, rangeEnd: end } = context;
+    const segmentId = Segment.getSegmentLocalId(context.url, {
+      start,
+      end,
+    });
+    const playlist = this.segmentManager.getPlaylistBySegmentId(segmentId);
+    this.debug(
+      "downloaded segment from playlist\n",
+      `playlist v: ${playlist?.index}\n`,
+      `segment: `,
+      playlist?.segments.get(segmentId)?.index
+    );
+    return playlist;
+  }
+
+  async fetchSegment(segmentUrl: string, byteRange?: ByteRange) {
+    const headers = new Headers();
+
+    if (byteRange) {
+      const { start, end } = byteRange;
+      const byteRangeString = `bytes=${start}-${end}`;
+      headers.set("Range", byteRangeString);
+    }
+    const response = await fetch(segmentUrl, {
+      headers,
+      signal: this.abortController.signal,
     });
+    if (!response.ok) {
+      throw new FetchError(
+        response.statusText ?? "Fetch, bad network response",
+        response.status,
+        response
+      );
+    }
+    const data = await response.arrayBuffer();
+    return {
+      ok: response.ok,
+      status: response.status,
+      data,
+      url: response.url,
+    };
+  }
+
+  private handleError(thrownError: unknown) {
+    const error = { code: 0, text: "" };
+    let details: object | null = null;
+    if (thrownError instanceof FetchError) {
+      error.code = thrownError.code;
+      error.text = thrownError.message;
+      details = thrownError.details;
+    } else if (thrownError instanceof Error) {
+      error.text = thrownError.message;
+    }
+    this.callbacks?.onError(error, this.context, details, this.stats);
+  }
+
+  private abortInternal() {
+    if (!this.response?.ok) {
+      this.abortController.abort();
+      this.stats.aborted = true;
+    }
   }
 
   abort() {
-    this.defaultLoader.abort();
+    if (this.defaultLoader) {
+      this.defaultLoader.abort();
+    } else {
+      this.abortInternal();
+      this.callbacks?.onAbort?.(this.stats, this.context, {});
+    }
   }
 
   destroy() {
-    this.defaultLoader.destroy();
+    if (this.defaultLoader) {
+      this.defaultLoader.destroy();
+    } else {
+      this.abortInternal();
+      this.callbacks = null;
+      this.config = null;
+    }
+  }
+}
+
+function getByteRange(
+  start: number | undefined,
+  end: number | undefined
+): ByteRange | undefined {
+  if (start !== undefined && end !== undefined && end > start) {
+    return { start, end: end - 1 };
+  }
+}
+
+function getLoadingStat({
+  loadedBytes,
+  targetBitrate,
+  loadingEndTime,
+}: {
+  targetBitrate: number;
+  loadedBytes: number;
+  loadingEndTime: number;
+}) {
+  const bits = loadedBytes * 8;
+  const timeForLoading = (bits / targetBitrate) * 1000;
+  const first = loadingEndTime - timeForLoading;
+  const start = first - DEFAULT_DOWNLOAD_LATENCY;
+
+  return { start, first, end: loadingEndTime };
+}
+
+class FetchError extends Error {
+  public code: number;
+  public details: object;
+
+  constructor(message: string, code: number, details: object) {
+    super(message);
+    this.code = code;
+    this.details = details;
   }
 }

packages/p2p-media-loader-hlsjs/src/services/segment-mananger.ts

@@ -1,5 +1,9 @@
 import { Playlist, Segment } from "./playlist";
-import type { LevelLoadedData, ManifestLoadedData } from "hls.js";
+import type {
+  ManifestLoadedData,
+  LevelUpdatedData,
+  AudioTrackLoadedData,
+} from "hls.js";
 
 export class SegmentManager {
   playlists: Map<string, Playlist> = new Map();
@@ -16,23 +20,33 @@ export class SegmentManager {
       if (this.playlists.has(level.url)) return;
       this.playlists.set(
         level.url,
-        new Playlist({ type: "video", index, masterManifestUrl: url })
+        new Playlist({
+          type: "video",
+          index,
+          masterManifestUrl: url,
+        })
       );
     });
 
     audioTracks.forEach((track, index) => {
       if (this.playlists.has(track.url)) return;
       this.playlists.set(
         track.url,
-        new Playlist({ type: "audio", index, masterManifestUrl: url })
+        new Playlist({
+          type: "audio",
+          index,
+          masterManifestUrl: url,
+        })
       );
     });
   }
 
-  setPlaylist(data: LevelLoadedData) {
+  updatePlaylist(data: LevelUpdatedData | AudioTrackLoadedData) {
+    if (!data.details) return;
     const {
       details: { url, fragments, live },
     } = data;
+
     const playlist = this.playlists.get(url);
     if (!playlist) return;