Merge pull request #1592 from canalplus/misc/mandatory-clear-signal

[Proposal] Make SharedReference's clearSignal param mandatory

src/core/main/worker/worker_main.ts

@@ -523,23 +523,26 @@ function loadOrReloadPreparedContent(
     segmentQueueCreator,
   } = preparedContent;
   const { drmSystemId, enableFastSwitching, initialTime, onCodecSwitch } = val;
-  playbackObservationRef.onUpdate((observation) => {
-    if (preparedContent.decipherabilityFreezeDetector.needToReload(observation)) {
-      handleMediaSourceReload({
-        timeOffset: 0,
-        minimumPosition: 0,
-        maximumPosition: Infinity,
-      });
-    }
-
-    // Synchronize SegmentSinks with what has been buffered.
-    ["video" as const, "audio" as const, "text" as const].forEach((tType) => {
-      const segmentSinkStatus = segmentSinksStore.getStatus(tType);
-      if (segmentSinkStatus.type === "initialized") {
-        segmentSinkStatus.value.synchronizeInventory(observation.buffered[tType] ?? []);
+  playbackObservationRef.onUpdate(
+    (observation) => {
+      if (preparedContent.decipherabilityFreezeDetector.needToReload(observation)) {
+        handleMediaSourceReload({
+          timeOffset: 0,
+          minimumPosition: 0,
+          maximumPosition: Infinity,
+        });
       }
-    });
-  });
+
+      // Synchronize SegmentSinks with what has been buffered.
+      ["video" as const, "audio" as const, "text" as const].forEach((tType) => {
+        const segmentSinkStatus = segmentSinksStore.getStatus(tType);
+        if (segmentSinkStatus.type === "initialized") {
+          segmentSinkStatus.value.synchronizeInventory(observation.buffered[tType] ?? []);
+        }
+      });
+    },
+    { clearSignal: currentLoadCanceller.signal },
+  );
 
   const initialPeriod =
     manifest.getPeriodForTime(initialTime) ?? manifest.getNextPeriod(initialTime);

src/main_thread/init/multi_thread_content_initializer.ts

@@ -901,46 +901,52 @@ export default class MultiThreadContentInitializer extends ContentInitializer {
           const ref = new SharedReference<IAdaptationChoice | null | undefined>(
             undefined,
           );
-          ref.onUpdate((adapChoice) => {
-            if (this._currentContentInfo === null) {
-              ref.finish();
-              return;
-            }
-            if (!isNullOrUndefined(adapChoice)) {
-              adapChoice.representations.onUpdate((repChoice, stopListening) => {
-                if (this._currentContentInfo === null) {
-                  stopListening();
-                  return;
-                }
-                sendMessage(this._settings.worker, {
-                  type: MainThreadMessageType.RepresentationUpdate,
-                  contentId: this._currentContentInfo.contentId,
-                  value: {
-                    periodId: msgData.value.periodId,
-                    adaptationId: adapChoice.adaptationId,
-                    bufferType: msgData.value.bufferType,
-                    choice: repChoice,
+          ref.onUpdate(
+            (adapChoice) => {
+              if (this._currentContentInfo === null) {
+                ref.finish();
+                return;
+              }
+              if (!isNullOrUndefined(adapChoice)) {
+                adapChoice.representations.onUpdate(
+                  (repChoice, stopListening) => {
+                    if (this._currentContentInfo === null) {
+                      stopListening();
+                      return;
+                    }
+                    sendMessage(this._settings.worker, {
+                      type: MainThreadMessageType.RepresentationUpdate,
+                      contentId: this._currentContentInfo.contentId,
+                      value: {
+                        periodId: msgData.value.periodId,
+                        adaptationId: adapChoice.adaptationId,
+                        bufferType: msgData.value.bufferType,
+                        choice: repChoice,
+                      },
+                    });
                   },
-                });
+                  { clearSignal: this._initCanceller.signal },
+                );
+              }
+              sendMessage(this._settings.worker, {
+                type: MainThreadMessageType.TrackUpdate,
+                contentId: this._currentContentInfo.contentId,
+                value: {
+                  periodId: msgData.value.periodId,
+                  bufferType: msgData.value.bufferType,
+                  choice: isNullOrUndefined(adapChoice)
+                    ? adapChoice
+                    : {
+                        adaptationId: adapChoice.adaptationId,
+                        switchingMode: adapChoice.switchingMode,
+                        initialRepresentations: adapChoice.representations.getValue(),
+                        relativeResumingPosition: adapChoice.relativeResumingPosition,
+                      },
+                },
               });
-            }
-            sendMessage(this._settings.worker, {
-              type: MainThreadMessageType.TrackUpdate,
-              contentId: this._currentContentInfo.contentId,
-              value: {
-                periodId: msgData.value.periodId,
-                bufferType: msgData.value.bufferType,
-                choice: isNullOrUndefined(adapChoice)
-                  ? adapChoice
-                  : {
-                      adaptationId: adapChoice.adaptationId,
-                      switchingMode: adapChoice.switchingMode,
-                      initialRepresentations: adapChoice.representations.getValue(),
-                      relativeResumingPosition: adapChoice.relativeResumingPosition,
-                    },
-              },
-            });
-          });
+            },
+            { clearSignal: this._initCanceller.signal },
+          );
           this.trigger("periodStreamReady", {
             period,
             type: msgData.value.bufferType,

src/playback_observer/media_element_playback_observer.ts

@@ -258,25 +258,25 @@ export default class PlaybackObserver {
   /**
    * Register a callback so it regularly receives playback observations.
    * @param {Function} cb
-   * @param {Object} options - Configuration options:
+   * @param {Object} params - Configuration parameters:
    *   - `includeLastObservation`: If set to `true` the last observation will
    *     be first emitted synchronously.
    *   - `clearSignal`: If set, the callback will be unregistered when this
    *     CancellationSignal emits.
    */
   public listen(
     cb: (observation: IPlaybackObservation, stopListening: () => void) => void,
-    options?: {
+    params: {
       includeLastObservation?: boolean | undefined;
-      clearSignal?: CancellationSignal | undefined;
+      clearSignal: CancellationSignal;
     },
   ) {
-    if (this._canceller.isUsed() || options?.clearSignal?.isCancelled() === true) {
+    if (this._canceller.isUsed() || params.clearSignal.isCancelled()) {
       return noop;
     }
     this._observationRef.onUpdate(cb, {
-      clearSignal: options?.clearSignal,
-      emitCurrentValue: options?.includeLastObservation,
+      clearSignal: params.clearSignal,
+      emitCurrentValue: params.includeLastObservation,
     });
   }
 

src/playback_observer/types.ts

@@ -216,9 +216,9 @@ export interface IReadOnlyPlaybackObserver<TObservationType> {
    */
   listen(
     cb: (observation: TObservationType, stopListening: () => void) => void,
-    options?: {
+    options: {
       includeLastObservation?: boolean | undefined;
-      clearSignal?: CancellationSignal | undefined;
+      clearSignal: CancellationSignal;
     },
   ): void;
   /**

src/playback_observer/utils/generate_read_only_observer.ts

@@ -36,20 +36,17 @@ export default function generateReadOnlyObserver<TSource, TDest>(
     },
     listen(
       cb: (observation: TDest, stopListening: () => void) => void,
-      options?: {
+      params: {
         includeLastObservation?: boolean | undefined;
-        clearSignal?: CancellationSignal | undefined;
+        clearSignal: CancellationSignal;
       },
     ): void {
-      if (
-        cancellationSignal.isCancelled() ||
-        options?.clearSignal?.isCancelled() === true
-      ) {
+      if (cancellationSignal.isCancelled() || params.clearSignal.isCancelled()) {
         return;
       }
       mappedRef.onUpdate(cb, {
-        clearSignal: options?.clearSignal,
-        emitCurrentValue: options?.includeLastObservation,
+        clearSignal: params.clearSignal,
+        emitCurrentValue: params.includeLastObservation,
       });
     },
     deriveReadOnlyObserver<TNext>(

src/playback_observer/worker_playback_observer.ts

@@ -111,21 +111,18 @@ export default class WorkerPlaybackObserver
 
   public listen(
     cb: (observation: IWorkerPlaybackObservation, stopListening: () => void) => void,
-    options?: {
+    params: {
       includeLastObservation?: boolean | undefined;
-      clearSignal?: CancellationSignal | undefined;
+      clearSignal: CancellationSignal;
     },
   ): void {
-    if (
-      this._cancelSignal.isCancelled() ||
-      options?.clearSignal?.isCancelled() === true
-    ) {
+    if (this._cancelSignal.isCancelled() || params.clearSignal.isCancelled()) {
       return;
     }
 
     this._src.onUpdate(cb, {
-      clearSignal: options?.clearSignal,
-      emitCurrentValue: options?.includeLastObservation,
+      clearSignal: params.clearSignal,
+      emitCurrentValue: params.includeLastObservation,
     });
   }
 

src/utils/reference.ts

@@ -174,24 +174,22 @@ class SharedReference<T> {
    * @param {Function} cb - Callback to be called each time the reference is
    * updated. Takes as first argument its new value and in second argument a
    * callback allowing to unregister the callback.
-   * @param {Object|undefined} [options]
-   * @param {Object|undefined} [options.clearSignal] - Allows to provide a
-   * CancellationSignal which will unregister the callback when it emits.
-   * @param {boolean|undefined} [options.emitCurrentValue] - If `true`, the
+   * @param {Object} params
+   * @param {Object} params.clearSignal - Allows to provide a CancellationSignal
+   * which will unregister the callback when it emits.
+   * @param {boolean|undefined} [params.emitCurrentValue] - If `true`, the
    * callback will also be immediately called with the current value.
    */
   public onUpdate(
     cb: (val: T, stopListening: () => void) => void,
-    options?:
-      | {
-          clearSignal?: CancellationSignal | undefined;
-          emitCurrentValue?: boolean | undefined;
-        }
-      | undefined,
+    params: {
+      clearSignal: CancellationSignal;
+      emitCurrentValue?: boolean | undefined;
+    },
   ): void {
     const unlisten = (): void => {
-      if (options?.clearSignal !== undefined) {
-        options.clearSignal.deregister(unlisten);
+      if (params.clearSignal !== undefined) {
+        params.clearSignal.deregister(unlisten);
       }
       if (cbObj.hasBeenCleared) {
         return;
@@ -206,18 +204,15 @@ class SharedReference<T> {
     const cbObj = { trigger: cb, complete: unlisten, hasBeenCleared: false };
     this._listeners.push(cbObj);
 
-    if (options?.emitCurrentValue === true) {
+    if (params.emitCurrentValue === true) {
       cb(this._value, unlisten);
     }
 
     if (this._isFinished || cbObj.hasBeenCleared) {
       unlisten();
       return;
     }
-    if (options?.clearSignal === undefined) {
-      return;
-    }
-    options.clearSignal.register(unlisten);
+    params.clearSignal.register(unlisten);
   }
 
   /**
@@ -240,13 +235,13 @@ class SharedReference<T> {
    * ```
    * @param {Function} cb - Callback to be called each time the reference is
    * updated. Takes the new value in argument.
-   * @param {Object | undefined} [options]
-   * @param {Object | undefined} [options.clearSignal] - Allows to provide a
+   * @param {Object} params
+   * @param {Object} params.clearSignal - Allows to provide a
    * CancellationSignal which will unregister the callback when it emits.
    */
   public waitUntilDefined(
     cb: (val: Exclude<T, undefined>) => void,
-    options?: { clearSignal?: CancellationSignal | undefined } | undefined,
+    params: { clearSignal: CancellationSignal },
   ): void {
     this.onUpdate(
       (val: T, stopListening) => {
@@ -255,7 +250,7 @@ class SharedReference<T> {
           cb(this._value as Exclude<T, undefined>);
         }
       },
-      { clearSignal: options?.clearSignal, emitCurrentValue: true },
+      { clearSignal: params.clearSignal, emitCurrentValue: true },
     );
   }