Requests container with p2p integration.

packages/p2p-media-loader-core/src/core.ts

@@ -15,9 +15,9 @@ export class Core<TStream extends Stream = Stream> {
   private readonly streams = new Map<string, StreamWithSegments<TStream>>();
   private readonly settings: Settings = {
     simultaneousHttpDownloads: 3,
-    highDemandBufferLength: 25,
-    httpBufferLength: 60,
-    p2pBufferLength: 60,
+    highDemandTimeWindow: 25,
+    httpDownloadTimeWindow: 60,
+    p2pDownloadTimeWindow: 60,
     cachedSegmentExpiration: 120,
     cachedSegmentsCount: 50,
   };
@@ -76,12 +76,12 @@ export class Core<TStream extends Stream = Stream> {
         new HybridLoader(this.settings, this.bandwidthApproximator);
       loader = this.secondaryStreamLoader;
     }
-    return loader.loadSegment(segment, stream);
+    return loader.loadSegmentByEngine(segment, stream);
   }
 
   abortSegmentLoading(segmentId: string): void {
-    this.mainStreamLoader.abortSegment(segmentId);
-    this.secondaryStreamLoader?.abortSegment(segmentId);
+    this.mainStreamLoader.abortSegmentByEngine(segmentId);
+    this.secondaryStreamLoader?.abortSegmentByEngine(segmentId);
   }
 
   updatePlayback(position: number, rate: number): void {

packages/p2p-media-loader-core/src/errors.ts

@@ -8,3 +8,9 @@ export class FetchError extends Error {
     this.details = details;
   }
 }
+
+export class AbortError extends Error {
+  constructor(message = "AbortError") {
+    super(message);
+  }
+}

packages/p2p-media-loader-core/src/http-loader.ts

@@ -1,76 +1,48 @@
-import { FetchError } from "./errors";
+import { AbortError, FetchError } from "./errors";
 import { Segment } from "./types";
+import { HttpRequest } from "./request";
+
+export function loadSegmentThroughHttp(
+  segment: Segment
+): Readonly<HttpRequest> {
+  const { promise, abortController } = fetchSegment(segment);
+  return {
+    type: "http",
+    promise,
+    abort: () => abortController.abort(),
+  };
+}
 
-type Request = {
-  promise: Promise<ArrayBuffer>;
-  abortController: AbortController;
-};
-
-export class HttpLoader {
-  private readonly requests = new Map<string, Request>();
+function fetchSegment(segment: Segment) {
+  const headers = new Headers();
+  const { url, byteRange, localId: segmentId } = segment;
 
-  async load(segment: Segment) {
-    const abortController = new AbortController();
-    const promise = this.fetch(segment, abortController);
-    const requestContext: Request = {
-      abortController,
-      promise,
-    };
-    this.requests.set(segment.localId, requestContext);
-    await promise;
-    this.requests.delete(segment.localId);
-    return promise;
+  if (byteRange) {
+    const { start, end } = byteRange;
+    const byteRangeString = `bytes=${start}-${end}`;
+    headers.set("Range", byteRangeString);
   }
+  const abortController = new AbortController();
 
-  private async fetch(segment: Segment, abortController: AbortController) {
-    const headers = new Headers();
-    const { url, byteRange } = segment;
+  const promise = fetch(url, {
+    headers,
+    signal: abortController.signal,
+  })
+    .then((response) => {
+      if (response.ok) return response.arrayBuffer();
 
-    if (byteRange) {
-      const { start, end } = byteRange;
-      const byteRangeString = `bytes=${start}-${end}`;
-      headers.set("Range", byteRangeString);
-    }
-    const response = await fetch(url, {
-      headers,
-      signal: abortController.signal,
-    });
-    if (!response.ok) {
       throw new FetchError(
-        response.statusText ?? "Fetch, bad network response",
+        response.statusText ?? `Network response was not for ${segmentId}`,
         response.status,
         response
       );
-    }
-
-    return response.arrayBuffer();
-  }
-
-  isLoading(segmentId: string) {
-    return this.requests.has(segmentId);
-  }
-
-  abort(segmentId: string) {
-    this.requests.get(segmentId)?.abortController.abort();
-    this.requests.delete(segmentId);
-  }
-
-  getLoadingsAmount() {
-    return this.requests.size;
-  }
-
-  getLoadingSegmentIds() {
-    return this.requests.keys();
-  }
-
-  getRequest(segmentId: string) {
-    return this.requests.get(segmentId)?.promise;
-  }
+    })
+    .catch((error) => {
+      if (error instanceof Error && error.name === "AbortError") {
+        throw new AbortError(`Segment fetch was aborted ${segmentId}`);
+      }
+      throw error;
+    });
 
-  abortAll() {
-    for (const request of this.requests.values()) {
-      request.abortController.abort();
-    }
-    this.requests.clear();
-  }
+  return { promise, abortController };
 }

packages/p2p-media-loader-core/src/hybrid-loader.ts

@@ -1,17 +1,18 @@
 import { Segment, SegmentResponse, StreamWithSegments } from "./index";
-import { HttpLoader } from "./http-loader";
+import { loadSegmentThroughHttp } from "./http-loader";
 import { P2PLoader } from "./p2p-loader";
 import { SegmentsMemoryStorage } from "./segments-storage";
 import { Settings } from "./types";
 import { BandwidthApproximator } from "./bandwidth-approximator";
 import { Playback, QueueItem } from "./internal-types";
+import { RequestContainer } from "./request";
 import * as Utils from "./utils";
+import { FetchError } from "./errors";
 
 export class HybridLoader {
   private streamManifestUrl?: string;
-  private readonly httpLoader = new HttpLoader();
+  private readonly requests = new RequestContainer();
   private p2pLoader?: P2PLoader;
-  private readonly pluginRequests = new Map<string, Request>();
   private readonly segmentStorage: SegmentsMemoryStorage;
   private storageCleanUpIntervalId?: number;
   private activeStream?: Readonly<StreamWithSegments>;
@@ -50,11 +51,12 @@ export class HybridLoader {
     this.p2pLoader = new P2PLoader(
       this.streamManifestUrl,
       stream,
+      this.requests,
       this.segmentStorage
     );
   }
 
-  async loadSegment(
+  async loadSegmentByEngine(
     segment: Readonly<Segment>,
     stream: Readonly<StreamWithSegments>
   ): Promise<SegmentResponse> {
@@ -68,15 +70,18 @@ export class HybridLoader {
     this.lastRequestedSegment = segment;
     void this.processQueue();
 
-    const storageData = await this.segmentStorage.getSegment(segment.localId);
+    const storageData = await this.segmentStorage.getSegmentData(
+      segment.localId
+    );
     if (storageData) {
       return {
         data: storageData,
         bandwidth: this.bandwidthApproximator.getBandwidth(),
       };
     }
-    const request = this.createPluginSegmentRequest(segment);
-    return request.responsePromise;
+    const request = Utils.getControlledPromise<SegmentResponse>();
+    this.requests.addEngineRequest(segment, request);
+    return request.promise;
   }
 
   private async processQueue(force = true) {
@@ -102,76 +107,67 @@ export class HybridLoader {
       isSegmentLoaded: (segmentId) => storedSegmentIds.has(segmentId),
     });
 
-    const bufferRanges = Utils.getLoadBufferRanges(
-      this.playback,
-      this.settings
+    this.requests.abortAllNotRequestedByEngine((segmentId) =>
+      queueSegmentIds.has(segmentId)
     );
-    for (const segmentId of this.getLoadingSegmentIds()) {
-      const segment = this.activeStream.segments.get(segmentId);
-      if (
-        !queueSegmentIds.has(segmentId) &&
-        !this.pluginRequests.has(segmentId) &&
-        !(segment && Utils.isSegmentActual(segment, bufferRanges))
-      ) {
-        this.abortSegment(segmentId);
-      }
-    }
 
     const { simultaneousHttpDownloads } = this.settings;
     for (const { segment, statuses } of queue) {
-      if (this.httpLoader.isLoading(segment.localId)) continue;
+      if (this.requests.isHttpRequested(segment.localId)) continue;
       if (statuses.has("high-demand")) {
-        if (this.httpLoader.getLoadingsAmount() < simultaneousHttpDownloads) {
+        if (this.requests.countHttpRequests() < simultaneousHttpDownloads) {
           void this.loadSegmentThroughHttp(segment);
           continue;
         }
         this.abortLastHttpLoadingAfter(queue, segment.localId);
-        if (this.httpLoader.getLoadingsAmount() < simultaneousHttpDownloads) {
+        if (this.requests.countHttpRequests() < simultaneousHttpDownloads) {
           void this.loadSegmentThroughHttp(segment);
         }
       }
       break;
     }
   }
 
-  getLoadingSegmentIds() {
-    return this.httpLoader.getLoadingSegmentIds();
-  }
-
-  abortSegment(segmentId: string) {
-    this.httpLoader.abort(segmentId);
-    const request = this.pluginRequests.get(segmentId);
-    if (!request) return;
-    request.onError("Abort");
-    this.pluginRequests.delete(segmentId);
+  abortSegmentByEngine(segmentId: string) {
+    this.requests.abortEngineRequest(segmentId);
   }
 
   private async loadSegmentThroughHttp(segment: Segment) {
     let data: ArrayBuffer | undefined;
     try {
-      data = await this.httpLoader.load(segment);
+      const httpRequest = loadSegmentThroughHttp(segment);
+      this.requests.addLoaderRequest(segment, httpRequest);
+      data = await httpRequest.promise;
     } catch (err) {
-      // TODO: handle abort
+      if (err instanceof FetchError) {
+        // TODO: handle error
+      }
     }
-    if (!data) return;
+    if (data) this.handleSegmentLoaded(segment, data);
+  }
+
+  private async loadThroughP2P(segment: Segment) {
+    if (!this.p2pLoader) return;
+    const data = await this.p2pLoader.downloadSegment(segment);
+    if (data) this.handleSegmentLoaded(segment, data);
+  }
+
+  private handleSegmentLoaded(segment: Segment, data: ArrayBuffer) {
     this.bandwidthApproximator.addBytes(data.byteLength);
     void this.segmentStorage.storeSegment(segment, data);
-    const request = this.pluginRequests.get(segment.localId);
-    if (request) {
-      request.onSuccess({
-        bandwidth: this.bandwidthApproximator.getBandwidth(),
-        data,
-      });
-    }
-    this.pluginRequests.delete(segment.localId);
+    this.requests.resolveEngineRequest(segment.localId, {
+      data,
+      bandwidth: this.bandwidthApproximator.getBandwidth(),
+    });
   }
 
   private abortLastHttpLoadingAfter(queue: QueueItem[], segmentId: string) {
-    for (let i = queue.length - 1; i >= 0; i--) {
-      const { segment } = queue[i];
-      if (segment.localId === segmentId) break;
-      if (this.httpLoader.isLoading(segment.localId)) {
-        this.abortSegment(segment.localId);
+    for (const {
+      segment: { localId: queueSegmentId },
+    } of arrayBackwards(queue)) {
+      if (queueSegmentId === segmentId) break;
+      if (this.requests.isHttpRequested(queueSegmentId)) {
+        this.requests.abortLoaderRequest(queueSegmentId);
         break;
       }
     }
@@ -189,40 +185,17 @@ export class HybridLoader {
     void this.processQueue(false);
   }
 
-  private createPluginSegmentRequest(segment: Segment) {
-    let onSuccess: Request["onSuccess"];
-    let onError: Request["onError"];
-    const responsePromise = new Promise<SegmentResponse>((resolve, reject) => {
-      onSuccess = resolve;
-      onError = reject;
-    });
-    const request: Request = {
-      responsePromise,
-      // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
-      onSuccess: onSuccess!,
-      // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
-      onError: onError!,
-    };
-
-    this.pluginRequests.set(segment.localId, request);
-    return request;
-  }
-
   destroy() {
     clearInterval(this.storageCleanUpIntervalId);
     this.storageCleanUpIntervalId = undefined;
     void this.segmentStorage.destroy();
-    this.httpLoader.abortAll();
-    for (const request of this.pluginRequests.values()) {
-      request.onError("Aborted");
-    }
-    this.pluginRequests.clear();
+    this.requests.destroy();
     this.playback = undefined;
   }
 }
 
-type Request = {
-  responsePromise: Promise<SegmentResponse>;
-  onSuccess: (response: SegmentResponse) => void;
-  onError: (reason?: unknown) => void;
-};
+function* arrayBackwards<T>(arr: T[]) {
+  for (let i = arr.length - 1; i >= 0; i--) {
+    yield arr[i];
+  }
+}