Load controlling: pure functions style.

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

@@ -64,13 +64,6 @@ export class Core<TStream extends Stream = Stream> {
 
     addSegments?.forEach((s) => stream.segments.addToEnd(s.localId, s));
     removeSegmentIds?.forEach((id) => stream.segments.delete(id));
-
-    const firstSegment = stream.segments.first?.[1];
-    if (firstSegment && firstSegment.startTime > this.position) {
-      this.position = firstSegment.startTime;
-      this.mainStreamLoader.updatePlayback(firstSegment.startTime);
-      this.secondaryStreamLoader.updatePlayback(firstSegment.startTime);
-    }
   }
 
   loadSegment(segmentLocalId: string): Promise<SegmentResponse> {
@@ -91,6 +84,8 @@ export class Core<TStream extends Stream = Stream> {
   updatePlayback(position: number, rate: number): void {
     this.mainStreamLoader.updatePlayback(position, rate);
     this.secondaryStreamLoader.updatePlayback(position, rate);
+
+    // TODO: update playback position when the live stream is updated
   }
 
   destroy(): void {

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

@@ -59,6 +59,10 @@ export class HttpLoader {
     return this.requests.size;
   }
 
+  getLoadingSegmentIds() {
+    return this.requests.keys();
+  }
+
   getRequest(segmentId: string) {
     return this.requests.get(segmentId)?.promise;
   }

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

@@ -1,35 +1,35 @@
 import { Segment, SegmentResponse, StreamWithSegments } from "./index";
 import { HttpLoader } from "./http-loader";
-import { LoadQueue } from "./load-queue";
 import { SegmentsMemoryStorage } from "./segments-storage";
 import { Settings } from "./types";
-import { Playback } from "./playback";
-import * as Utils from "./utils";
 import { BandwidthApproximator } from "./bandwidth-approximator";
+import { Playback, QueueItem } from "./internal-types";
+import * as Utils from "./utils";
 
 export class HybridLoader {
-  private readonly queue: LoadQueue;
   private readonly httpLoader = new HttpLoader();
   private readonly pluginRequests = new Map<string, Request>();
   private readonly segmentStorage: SegmentsMemoryStorage;
   private storageCleanUpIntervalId?: number;
-  private readonly playback: Playback;
+  private activeStream?: Readonly<StreamWithSegments>;
+  private lastRequestedSegment?: Readonly<Segment>;
+  private playback?: Playback;
+  private segmentAvgLength?: number;
 
   constructor(
     private readonly settings: Settings,
     private readonly bandwidthApproximator: BandwidthApproximator
   ) {
     this.segmentStorage = new SegmentsMemoryStorage(this.settings);
-    this.playback = new Playback(this.settings);
-    this.queue = new LoadQueue(this.playback);
-    this.queue.subscribeToUpdate(this.onQueueUpdated.bind(this));
-    this.queue.setIsSegmentLoadedPredicate(this.isSegmentLoaded.bind(this));
     this.segmentStorage.setIsSegmentLockedPredicate((segment) => {
-      const stream = this.queue.activeStream;
-      return !!(
-        stream?.segments.has(segment.localId) &&
-        Utils.getSegmentLoadStatuses(segment, this.playback)
+      if (!this.playback || !this.activeStream?.segments.has(segment.localId)) {
+        return false;
+      }
+      const bufferRanges = Utils.getLoadBufferRanges(
+        this.playback,
+        this.settings
       );
+      return Utils.isSegmentActual(segment, bufferRanges);
     });
 
     this.storageCleanUpIntervalId = setInterval(
@@ -39,13 +39,19 @@ export class HybridLoader {
   }
 
   async loadSegment(
-    segment: Segment,
-    stream: StreamWithSegments
+    segment: Readonly<Segment>,
+    stream: Readonly<StreamWithSegments>
   ): Promise<SegmentResponse> {
-    if (!this.playback.isInitialized()) {
-      this.playback.position = segment.startTime;
+    if (!this.playback) {
+      this.playback = { position: segment.startTime, rate: 1 };
     }
-    this.queue.updateOnSegmentRequest(segment, stream);
+    if (stream !== this.activeStream) {
+      this.segmentAvgLength = computeSegmentAvgLength(stream);
+      this.activeStream = stream;
+    }
+    this.lastRequestedSegment = segment;
+    this.processQueue();
+
     const storageData = await this.segmentStorage.getSegment(segment.localId);
     if (storageData) {
       return {
@@ -57,20 +63,43 @@ export class HybridLoader {
     return request.responsePromise;
   }
 
-  abortSegment(segmentId: string) {
-    this.httpLoader.abort(segmentId);
-  }
-
   private processQueue() {
+    if (!this.activeStream || !this.lastRequestedSegment || !this.playback) {
+      return;
+    }
+
+    const { queue, queueSegmentIds } = Utils.generateQueue({
+      segment: this.lastRequestedSegment,
+      stream: this.activeStream,
+      playback: this.playback,
+      settings: this.settings,
+      isSegmentLoaded: (segmentId) => this.segmentStorage.has(segmentId),
+    });
+
+    const bufferRanges = Utils.getLoadBufferRanges(
+      this.playback,
+      this.settings
+    );
+    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 this.queue.items()) {
+    for (const { segment, statuses } of queue) {
       if (this.httpLoader.isLoading(segment.localId)) continue;
       if (statuses.has("high-demand")) {
         if (this.httpLoader.getLoadingsAmount() < simultaneousHttpDownloads) {
           void this.loadSegmentThroughHttp(segment);
           continue;
         }
-        this.abortLastHttpLoadingAfter(segment.localId);
+        this.abortLastHttpLoadingAfter(queue, segment.localId);
         if (this.httpLoader.getLoadingsAmount() < simultaneousHttpDownloads) {
           void this.loadSegmentThroughHttp(segment);
         }
@@ -79,11 +108,28 @@ export class HybridLoader {
     }
   }
 
+  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);
+  }
+
   private async loadSegmentThroughHttp(segment: Segment) {
-    const data = await this.httpLoader.load(segment);
+    let data: ArrayBuffer | undefined;
+    try {
+      data = await this.httpLoader.load(segment);
+    } catch (err) {
+      // TODO: handle abort
+    }
+    if (!data) return;
     this.bandwidthApproximator.addBytes(data.byteLength);
     void this.segmentStorage.storeSegment(segment, data);
-    this.queue.removeLoadedSegment(segment.localId);
     const request = this.pluginRequests.get(segment.localId);
     if (request) {
       request.onSuccess({
@@ -94,34 +140,30 @@ export class HybridLoader {
     this.pluginRequests.delete(segment.localId);
   }
 
-  private abortLastHttpLoadingAfter(segmentId: string) {
-    for (const { segment } of this.queue.itemsBackwards()) {
+  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.httpLoader.abort(segment.localId);
+        this.abortSegment(segment.localId);
         break;
       }
     }
   }
 
   updatePlayback(position: number, rate?: number) {
-    this.playback.position = position;
-    if (rate !== undefined) this.playback.rate = rate;
-    this.queue.updateOnPlaybackChange();
-  }
+    if (!this.playback) return;
+    const isRateChanged = rate && this.playback.rate !== rate;
+    const isPositionSignificantlyChanged =
+      this.segmentAvgLength === undefined ||
+      Math.abs(position - this.playback.position) / this.segmentAvgLength >=
+        0.45;
 
-  private onQueueUpdated(removedSegmentIds?: string[]) {
-    removedSegmentIds?.forEach((id) => {
-      this.httpLoader.abort(id);
-      const request = this.pluginRequests.get(id);
-      if (request) request.onError("aborted");
-      this.pluginRequests.delete(id);
-    });
-    this.processQueue();
-  }
+    if (!isRateChanged && !isPositionSignificantlyChanged) return;
 
-  private isSegmentLoaded(segmentId: string): boolean {
-    return this.segmentStorage.hasSegment(segmentId);
+    if (isPositionSignificantlyChanged) this.playback.position = position;
+    if (isRateChanged) this.playback.rate = rate;
+    this.processQueue();
   }
 
   private createPluginSegmentRequest(segment: Segment) {
@@ -144,7 +186,6 @@ export class HybridLoader {
   }
 
   clear() {
-    this.queue.clear();
     clearInterval(this.storageCleanUpIntervalId);
     this.storageCleanUpIntervalId = undefined;
     void this.segmentStorage.clear();
@@ -153,7 +194,7 @@ export class HybridLoader {
       request.onError("Aborted");
     }
     this.pluginRequests.clear();
-    this.playback.clear();
+    this.playback = undefined;
   }
 }
 
@@ -162,3 +203,12 @@ type Request = {
   onSuccess: (response: SegmentResponse) => void;
   onError: (reason?: unknown) => void;
 };
+
+function computeSegmentAvgLength(stream: StreamWithSegments) {
+  if (!stream.segments.size) return;
+  let sum = 0;
+  for (const segment of stream.segments.values()) {
+    sum += segment.endTime - segment.startTime;
+  }
+  return sum / stream.segments.size;
+}

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

@@ -1,3 +1,5 @@
+import { Segment } from "./types";
+
 export type Playback = {
   position: number;
   rate: number;
@@ -7,3 +9,16 @@ export type SegmentLoadStatus =
   | "high-demand"
   | "http-downloadable"
   | "p2p-downloadable";
+
+export type NumberRange = {
+  from: number;
+  to: number;
+};
+
+export type LoadBufferRanges = {
+  highDemand: NumberRange;
+  http: NumberRange;
+  p2p: NumberRange;
+};
+
+export type QueueItem = { segment: Segment; statuses: Set<SegmentLoadStatus> };

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

@@ -40,32 +40,6 @@ export class LinkedMap<K, V extends object> {
     this.map.set(key, item);
   }
 
-  addToStart(key: K, value: V) {
-    const item: LinkedObject<K, V> = { item: [key, value] };
-    if (this._first) {
-      this._first.prev = item;
-      item.next = this._first;
-    }
-    this._first = item;
-    if (!this._last) this._last = item;
-    this.map.set(key, item);
-  }
-
-  addAfter(prevKey: K, key: K, value: V) {
-    const prev = this.map.get(prevKey);
-    if (!prev) return;
-
-    const newItem: LinkedObject<K, V> = {
-      item: [key, value],
-      prev,
-      next: prev.next,
-    };
-    prev.next = newItem;
-    if (this._last === prev) this._last = newItem;
-
-    this.map.set(key, newItem);
-  }
-
   delete(key: K) {
     if (!this.map.size) return;
     const value = this.map.get(key);
@@ -85,27 +59,9 @@ export class LinkedMap<K, V extends object> {
     this.map.clear();
   }
 
-  *entriesBackwards(key?: K): Generator<[K, V]> {
-    let value = key ? this.map.get(key) : this._last;
-    if (value === undefined) return;
-    while (value?.item !== undefined) {
-      yield value.item;
-      value = value.prev;
-    }
-  }
-
-  *entries(key?: K): Generator<[K, V]> {
+  *values(key?: K) {
     let value = key ? this.map.get(key) : this._first;
     if (value === undefined) return;
-    while (value?.item !== undefined) {
-      yield value.item;
-      value = value.next;
-    }
-  }
-
-  *values() {
-    let value = this._first;
-    if (value === undefined) return;
     while (value?.item !== undefined) {
       yield value.item[1];
       value = value.next;
@@ -121,12 +77,6 @@ export class LinkedMap<K, V extends object> {
     }
   }
 
-  forEach(callback: (item: [K, V]) => void) {
-    for (const value of this.entries()) {
-      callback(value);
-    }
-  }
-
   getNextTo(key: K): [K, V] | undefined {
     return this.map.get(key)?.next?.item;
   }