mirror of
https://github.com/zoriya/v10.git
synced 2026-08-16 02:45:09 +00:00
feat(store): align queue with native (#312)
This commit is contained in:
rahim
@@ -0,0 +1,424 @@
|
||||
# Store Queue Design
|
||||
|
||||
## Overview
|
||||
|
||||
The `@videojs/store` queue manages request execution for media player state. This document describes how the native `<video>` element handles async operations internally, and how our queue mirrors and extends that behavior.
|
||||
|
||||
## Native Media Element Queue
|
||||
|
||||
The HTML spec doesn't expose a formal "queue" API, but the media element has internal queuing behavior managed through the browser's event loop.
|
||||
|
||||
### Media Element Event Task Source
|
||||
|
||||
The spec defines a **media element event task source** used when queuing async operations:
|
||||
|
||||
> "To queue a media element task with a media element element and a series of steps steps, queue an element task on the media element's media element event task source given element and steps."
|
||||
|
||||
- [WHATWG HTML Spec: Media Elements](https://html.spec.whatwg.org/multipage/media.html)
|
||||
|
||||
This means media operations are queued as **tasks** in the event loop, not microtasks. When you set `currentTime`, the actual seek and resulting events (`seeking`, `seeked`) happen asynchronously on subsequent event loop ticks.
|
||||
|
||||
### Pending Play Promises
|
||||
|
||||
The spec maintains a **list of pending play promises** on each media element:
|
||||
|
||||
```
|
||||
1. play() called → create promise, add to list
|
||||
2. play() called again → create another promise, add to same list
|
||||
3. Playback starts → resolve ALL promises in list
|
||||
4. (or) Playback aborted → reject ALL promises with AbortError
|
||||
```
|
||||
|
||||
This is documented in:
|
||||
|
||||
- [WHATWG HTML Spec: play() method](https://html.spec.whatwg.org/multipage/media.html#dom-media-play)
|
||||
- [MDN: HTMLMediaElement.load()](https://developer.mozilla.org/en-US/docs/Web/API/HTMLMediaElement/load)
|
||||
|
||||
Multiple `play()` calls don't queue multiple play operations — they share the fate of a single logical playback attempt.
|
||||
|
||||
### Load Algorithm
|
||||
|
||||
When `load()` is called, the media element:
|
||||
|
||||
1. Aborts any running resource selection algorithm
|
||||
2. Takes all pending tasks from the media element event task source
|
||||
3. **Immediately resolves or rejects pending play promises** (doesn't orphan them)
|
||||
4. Removes pending tasks from the queue
|
||||
5. Resets state (`paused = true`, `seeking = false`, `currentTime = 0`, etc.)
|
||||
|
||||
> "Basically, pending events and callbacks are discarded and promises in-flight to be resolved/rejected are resolved/rejected immediately when the media element starts loading a new resource."
|
||||
|
||||
- [WHATWG HTML Spec](https://html.spec.whatwg.org/multipage/media.html)
|
||||
|
||||
### Seek Behavior
|
||||
|
||||
Setting `currentTime` triggers the seeking algorithm:
|
||||
|
||||
- The property set returns immediately (sync)
|
||||
- The actual seek happens async
|
||||
- If you set `currentTime` again before `seeked` fires, the browser coalesces to the new target
|
||||
- You get `seeking` event(s), then a single `seeked` at the final position
|
||||
|
||||
The browser doesn't queue seeks — it's **last-write-wins**.
|
||||
|
||||
### Key Native Characteristics
|
||||
|
||||
| Behavior | Description |
|
||||
| ------------------------- | ------------------------------------------------------- |
|
||||
| **Lossy/Coalescing** | Same-type operations collapse (seeks, play calls) |
|
||||
| **Last-write-wins** | New value supersedes pending operation |
|
||||
| **Load as nuclear reset** | `load()` cancels everything, rejects pending promises |
|
||||
| **Ready-gating** | Operations wait for appropriate `readyState` |
|
||||
| **Async via task queue** | Operations execute on event loop, not synchronously |
|
||||
| **Play/pause interplay** | `pause()` during pending play rejects with `AbortError` |
|
||||
| **Shared fate for play** | Multiple `play()` calls share single outcome |
|
||||
| **No retries** | Failed operations don't retry automatically |
|
||||
| **No timeouts** | Operations can hang indefinitely |
|
||||
|
||||
## Our Queue Design
|
||||
|
||||
### Philosophy
|
||||
|
||||
We mirror native behavior where it makes sense, and layer observability where native falls short.
|
||||
|
||||
**Match native:**
|
||||
|
||||
- Lossy/coalescing via `key`
|
||||
- Last-write-wins via `SUPERSEDED`
|
||||
- Load as nuclear reset via `CANCEL_ALL`
|
||||
- Ready-gating via `guard`
|
||||
- Async execution
|
||||
- Play/pause coordination via shared `key`
|
||||
- No retries
|
||||
- No timeouts (but guards can timeout)
|
||||
|
||||
**Extend native:**
|
||||
|
||||
- Request identity (tasks have IDs)
|
||||
- Promise per request (native only has this for `play()`)
|
||||
- Observable status (`pending`, `success`, `error`)
|
||||
- Cancellation reasons (`SUPERSEDED`, `CANCELLED`, `ABORTED`, etc.)
|
||||
- Clean promise resolution (native orphans some promises)
|
||||
|
||||
### Sync Execution Model
|
||||
|
||||
Our queue doesn't actually "queue" in the traditional sense. When you call `enqueue()`:
|
||||
|
||||
1. **Immediately** abort any pending task with the same key (via `AbortController`)
|
||||
2. **Immediately** start executing the new task
|
||||
3. Return a promise that resolves/rejects when the task completes
|
||||
|
||||
This matches native behavior — `currentTime = x` starts the seek immediately, it doesn't wait for previous seeks to complete.
|
||||
|
||||
```ts
|
||||
// Native
|
||||
video.currentTime = 10; // starts immediately
|
||||
video.currentTime = 20; // starts immediately, browser drops first
|
||||
|
||||
// Our queue
|
||||
store.request.seek(10); // starts immediately
|
||||
store.request.seek(20); // starts immediately, aborts first via signal
|
||||
```
|
||||
|
||||
### Operation Comparison
|
||||
|
||||
#### Seek
|
||||
|
||||
**Native:**
|
||||
|
||||
```ts
|
||||
video.currentTime = 10;
|
||||
video.currentTime = 20;
|
||||
video.currentTime = 30;
|
||||
// Result: seeks to 30, intermediate seeks dropped
|
||||
// Events: seeking (maybe multiple), seeked (once, at 30)
|
||||
```
|
||||
|
||||
**Our queue:**
|
||||
|
||||
```ts
|
||||
const p1 = store.request.seek(10); // starts
|
||||
const p2 = store.request.seek(20); // aborts p1, starts
|
||||
const p3 = store.request.seek(30); // aborts p2, starts
|
||||
// p1 rejects SUPERSEDED
|
||||
// p2 rejects SUPERSEDED
|
||||
// p3 resolves when seeked
|
||||
```
|
||||
|
||||
✓ Same outcome, better observability.
|
||||
|
||||
#### Play then Pause
|
||||
|
||||
**Native:**
|
||||
|
||||
```ts
|
||||
const p = video.play();
|
||||
video.pause();
|
||||
// p rejects with AbortError
|
||||
// element is paused
|
||||
```
|
||||
|
||||
**Our queue:**
|
||||
|
||||
```ts
|
||||
const p = store.request.play();
|
||||
store.request.pause();
|
||||
// p rejects with SUPERSEDED
|
||||
// element is paused
|
||||
```
|
||||
|
||||
✓ Same outcome, different error type.
|
||||
|
||||
#### Multiple Play Calls
|
||||
|
||||
**Native:**
|
||||
|
||||
```ts
|
||||
const p1 = video.play();
|
||||
const p2 = video.play();
|
||||
const p3 = video.play();
|
||||
// Playback starts → p1, p2, p3 ALL resolve
|
||||
// Playback fails → p1, p2, p3 ALL reject
|
||||
```
|
||||
|
||||
**Our queue (current):**
|
||||
|
||||
```ts
|
||||
const p1 = store.request.play(); // starts
|
||||
const p2 = store.request.play(); // supersedes p1
|
||||
const p3 = store.request.play(); // supersedes p2
|
||||
// p1 rejects SUPERSEDED
|
||||
// p2 rejects SUPERSEDED
|
||||
// p3 resolves or rejects based on outcome
|
||||
```
|
||||
|
||||
⚠️ Different — we supersede, native shares fate.
|
||||
|
||||
**Our queue (with `mode: 'shared'`):**
|
||||
|
||||
```ts
|
||||
// play configured with mode: 'shared'
|
||||
const p1 = store.request.play(); // starts
|
||||
const p2 = store.request.play(); // joins p1
|
||||
const p3 = store.request.play(); // joins p1
|
||||
// All three resolve/reject together
|
||||
```
|
||||
|
||||
✓ Matches native with explicit opt-in.
|
||||
|
||||
#### Load
|
||||
|
||||
**Native:**
|
||||
|
||||
```ts
|
||||
video.play();
|
||||
video.load();
|
||||
// play promise rejects with AbortError
|
||||
// element resets completely
|
||||
```
|
||||
|
||||
**Our queue:**
|
||||
|
||||
```ts
|
||||
store.request.play();
|
||||
store.request.load(); // cancel: CANCEL_ALL
|
||||
// play promise rejects with CANCELLED
|
||||
// element resets
|
||||
```
|
||||
|
||||
✓ Same outcome, clean rejection.
|
||||
|
||||
### Request Configuration
|
||||
|
||||
#### Keys
|
||||
|
||||
Requests with the same `key` coordinate together. Default key is the request name.
|
||||
|
||||
```ts
|
||||
request: {
|
||||
play: {
|
||||
key: 'playback',
|
||||
handler: ...
|
||||
},
|
||||
pause: {
|
||||
key: 'playback', // same key — pause supersedes play
|
||||
handler: ...
|
||||
},
|
||||
seek: {
|
||||
// key defaults to 'seek'
|
||||
handler: ...
|
||||
},
|
||||
}
|
||||
```
|
||||
|
||||
#### Mode
|
||||
|
||||
The `mode` option controls how requests with the same key interact:
|
||||
|
||||
| Mode | Behavior | Use case |
|
||||
| ----------------------- | -------------------------------- | ------------------- |
|
||||
| `'exclusive'` (default) | Supersede pending request | seek, pause, volume |
|
||||
| `'shared'` | Join pending request, share fate | play |
|
||||
|
||||
```ts
|
||||
request: {
|
||||
play: {
|
||||
key: 'playback',
|
||||
mode: 'shared',
|
||||
handler: ...
|
||||
},
|
||||
pause: {
|
||||
key: 'playback',
|
||||
mode: 'exclusive', // default
|
||||
handler: ...
|
||||
},
|
||||
}
|
||||
```
|
||||
|
||||
This mirrors the Web Locks API naming:
|
||||
|
||||
- [MDN: Web Locks API](https://developer.mozilla.org/en-US/docs/Web/API/Web_Locks_API)
|
||||
|
||||
#### Cancel
|
||||
|
||||
The `cancel` option aborts other in-flight requests by name. Use `CANCEL_ALL` for nuclear reset (like `load()`).
|
||||
|
||||
```ts
|
||||
import { CANCEL_ALL } from '@videojs/store';
|
||||
|
||||
request: {
|
||||
load: {
|
||||
cancel: CANCEL_ALL, // cancels ALL pending requests
|
||||
handler: ...
|
||||
},
|
||||
stop: {
|
||||
cancel: ['seek', 'preload'], // cancels specific requests
|
||||
handler: ...
|
||||
},
|
||||
}
|
||||
```
|
||||
|
||||
#### Guards
|
||||
|
||||
Guards gate request execution, similar to how native operations wait for appropriate `readyState`.
|
||||
|
||||
```ts
|
||||
request: {
|
||||
seek: {
|
||||
guard: [hasMetadata],
|
||||
handler: (time, { target }) => {
|
||||
target.currentTime = time;
|
||||
},
|
||||
},
|
||||
play: {
|
||||
guard: timeout(canPlay, 5000),
|
||||
handler: ...
|
||||
},
|
||||
}
|
||||
```
|
||||
|
||||
Native has implicit ready-gating. Our guards make it explicit and configurable.
|
||||
|
||||
### API Placement: Store vs Queue
|
||||
|
||||
#### Queue Responsibilities
|
||||
|
||||
The queue is a general-purpose task executor:
|
||||
|
||||
- Task lifecycle (pending → success/error)
|
||||
- Key-based coordination (supersede, shared)
|
||||
- Abort via signal
|
||||
- Observable task state
|
||||
|
||||
```ts
|
||||
queue.enqueue({
|
||||
name: 'seek',
|
||||
key: 'seek',
|
||||
handler: async ({ signal }) => { ... },
|
||||
});
|
||||
```
|
||||
|
||||
#### Store Responsibilities
|
||||
|
||||
The store adds target-aware semantics:
|
||||
|
||||
- Target management (attach/detach)
|
||||
- State synchronization (getSnapshot, subscribe)
|
||||
- Guards (ready-gating)
|
||||
- Request metadata
|
||||
- Slice composition
|
||||
|
||||
```ts
|
||||
const store = createStore({
|
||||
slices: [playbackSlice],
|
||||
});
|
||||
|
||||
store.attach(videoElement);
|
||||
store.request.seek(30, { source: 'user' });
|
||||
```
|
||||
|
||||
#### Why Guards Live on Store
|
||||
|
||||
Guards are **target-aware**:
|
||||
|
||||
- They check `readyState`, `networkState`, target availability
|
||||
- They need access to the target (via slice context)
|
||||
- They're part of the request contract, not generic task execution
|
||||
|
||||
The queue doesn't know about targets. It just knows about tasks, keys, and abort signals. The store translates requests into queue tasks, including guard evaluation.
|
||||
|
||||
```ts
|
||||
// Store transforms this:
|
||||
store.request.seek(30);
|
||||
|
||||
// Into this queue task:
|
||||
queue.enqueue({
|
||||
name: 'seek',
|
||||
key: 'seek',
|
||||
handler: async (ctx) => {
|
||||
// Guards evaluated by store before this runs
|
||||
await seekHandler(30, ctx);
|
||||
},
|
||||
});
|
||||
```
|
||||
|
||||
### Error Codes
|
||||
|
||||
| Code | Description | Native Equivalent |
|
||||
| ------------ | ---------------------------------------------- | ----------------------------- |
|
||||
| `SUPERSEDED` | Replaced by same-key request | (implicit, no clean signal) |
|
||||
| `CANCELLED` | Cancelled by another request (`cancel: [...]`) | AbortError on load() |
|
||||
| `ABORTED` | Manually aborted via `queue.abort()` | AbortError |
|
||||
| `DESTROYED` | Store or queue destroyed | (n/a) |
|
||||
| `DETACHED` | Target detached | (n/a) |
|
||||
| `NO_TARGET` | No target attached | (implicit failure) |
|
||||
| `REJECTED` | Guard returned falsy | (implicit, operation ignored) |
|
||||
| `TIMEOUT` | Guard timed out | (native hangs forever) |
|
||||
|
||||
## Summary
|
||||
|
||||
Our queue mirrors native media element behavior:
|
||||
|
||||
- Sync property sets → immediate execution
|
||||
- Last-write-wins → key-based supersession
|
||||
- Shared fate for play → `mode: 'shared'`
|
||||
- Nuclear reset → `CANCEL_ALL`
|
||||
- Ready-gating → guards
|
||||
|
||||
And extends it with:
|
||||
|
||||
- Request identity
|
||||
- Promise per request
|
||||
- Observable status
|
||||
- Clean cancellation reasons
|
||||
|
||||
The goal is native-like performance with much better observability.
|
||||
|
||||
## References
|
||||
|
||||
- [WHATWG HTML Spec: Media Elements](https://html.spec.whatwg.org/multipage/media.html)
|
||||
- [MDN: HTMLMediaElement](https://developer.mozilla.org/en-US/docs/Web/API/HTMLMediaElement)
|
||||
- [MDN: HTMLMediaElement.load()](https://developer.mozilla.org/en-US/docs/Web/API/HTMLMediaElement/load)
|
||||
- [MDN: Web Locks API](https://developer.mozilla.org/en-US/docs/Web/API/Web_Locks_API)
|
||||
- [Chromium: play() Promise Implementation](https://groups.google.com/a/chromium.org/g/blink-reviews-html/c/ZIGTQ1yhJRo)
|
||||
- [WHATWG Issue #869: play() promise rejection](https://github.com/whatwg/html/issues/869)
|
||||
@@ -366,6 +366,23 @@ disconnect(): void {
|
||||
|
||||
For single cleanup, use a simple unsubscribe function.
|
||||
|
||||
### Promise Cleanup
|
||||
|
||||
Use `.finally()` for cleanup that runs regardless of success or failure:
|
||||
|
||||
```ts
|
||||
// Good - when awaiting or returning the promise
|
||||
await promise.finally(() => cache.delete(key));
|
||||
|
||||
// Good - fire-and-forget cleanup that shouldn't propagate rejection
|
||||
promise.then(
|
||||
() => cache.delete(key),
|
||||
() => cache.delete(key)
|
||||
);
|
||||
```
|
||||
|
||||
**Note:** `.finally()` propagates rejections to its returned promise. If you're not awaiting or returning it, use `.then()` with both handlers to avoid unhandled rejections.
|
||||
|
||||
### No Hungarian Type Notation
|
||||
|
||||
Never prefix type parameters with `T`. Use descriptive names instead:
|
||||
|
||||
@@ -154,9 +154,9 @@ request: {
|
||||
// Full config when needed
|
||||
play: {
|
||||
key: 'playback',
|
||||
mode: 'shared',
|
||||
guard: [],
|
||||
cancel: [],
|
||||
// ...
|
||||
async handler(_, { target, signal }) {
|
||||
target.play();
|
||||
await onEvent(target, 'play', { signal });
|
||||
@@ -332,17 +332,62 @@ request: {
|
||||
}
|
||||
```
|
||||
|
||||
### Mode
|
||||
|
||||
The `mode` option controls how requests with the same key interact:
|
||||
|
||||
| Mode | Behavior | Use case |
|
||||
| ----------------------- | -------------------------------- | ------------------- |
|
||||
| `'exclusive'` (default) | Supersede pending request | seek, pause, volume |
|
||||
| `'shared'` | Join pending request, share fate | play |
|
||||
|
||||
```ts
|
||||
request: {
|
||||
play: {
|
||||
key: 'playback',
|
||||
mode: 'shared', // Multiple play() calls share the same outcome
|
||||
async handler(_, { target }) {
|
||||
await target.play();
|
||||
},
|
||||
},
|
||||
pause: {
|
||||
key: 'playback',
|
||||
mode: 'exclusive', // default - supersedes play
|
||||
handler: (_, { target }) => target.pause(),
|
||||
},
|
||||
}
|
||||
```
|
||||
|
||||
With `mode: 'shared'`, multiple calls while a request is pending all resolve/reject together:
|
||||
|
||||
```ts
|
||||
const p1 = store.request.play(); // starts
|
||||
const p2 = store.request.play(); // joins p1
|
||||
const p3 = store.request.play(); // joins p1
|
||||
// All three resolve/reject together when playback starts or fails
|
||||
```
|
||||
|
||||
### Cancels
|
||||
|
||||
Requests can cancel other in-flight requests by name. Cancellation happens immediately when the
|
||||
request is enqueued, before guards.
|
||||
|
||||
```ts
|
||||
import { CANCEL_ALL } from '@videojs/store';
|
||||
|
||||
request: {
|
||||
stop: {
|
||||
cancel: ['seek', 'preload'], // Request names to cancel
|
||||
handler: (_, { target }) => target.pause(),
|
||||
},
|
||||
|
||||
load: {
|
||||
cancel: CANCEL_ALL, // Nuclear reset - cancels ALL pending requests
|
||||
handler: (src, { target }) => {
|
||||
target.src = src;
|
||||
target.load();
|
||||
},
|
||||
},
|
||||
}
|
||||
```
|
||||
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
import type { Request, RequestMeta } from './request';
|
||||
import type { Request, RequestMeta, RequestMode } from './request';
|
||||
import type { Reactive } from './state';
|
||||
import type { ErrorTask, PendingTask, SuccessTask, Task, TaskContext, TaskKey } from './task';
|
||||
|
||||
@@ -22,6 +22,7 @@ export type EnsureTaskRecord<T> = T extends TaskRecord ? T : never;
|
||||
export interface QueueTask<Key extends TaskKey = TaskKey, Input = unknown, Output = unknown> {
|
||||
name: string;
|
||||
key: Key;
|
||||
mode?: RequestMode;
|
||||
input?: Input;
|
||||
meta?: RequestMeta | null;
|
||||
handler: (ctx: TaskContext<Input>) => Promise<Output>;
|
||||
@@ -39,6 +40,8 @@ export class Queue<Tasks extends TaskRecord = DefaultTaskRecord> {
|
||||
/** Reactive tasks. Subscribe via `subscribe(queue.tasks, fn)`. */
|
||||
readonly tasks: Reactive<TasksRecord<Tasks>>;
|
||||
|
||||
readonly #sharedPromises = new Map<TaskKey, Promise<unknown>>();
|
||||
|
||||
#destroyed = false;
|
||||
|
||||
constructor() {
|
||||
@@ -70,20 +73,28 @@ export class Queue<Tasks extends TaskRecord = DefaultTaskRecord> {
|
||||
enqueue<K extends keyof Tasks>(
|
||||
task: QueueTask<TaskKey<K>, Tasks[K]['input'], Tasks[K]['output']>,
|
||||
): Promise<Tasks[K]['output']> {
|
||||
const { name, key, input, meta = null, handler } = task;
|
||||
const { name, key, mode = 'exclusive', input, meta = null, handler } = task;
|
||||
|
||||
if (this.#destroyed) {
|
||||
return Promise.reject(new StoreError('DESTROYED'));
|
||||
}
|
||||
|
||||
// Supersede any pending task with the same key (may have different name)
|
||||
for (const task of Object.values(this.tasks) as Task[]) {
|
||||
if (task?.key === key && task.status === 'pending') {
|
||||
task.abort.abort(new StoreError('SUPERSEDED'));
|
||||
// Shared mode: join existing pending task with same key
|
||||
if (mode === 'shared') {
|
||||
const existingPromise = this.#sharedPromises.get(key);
|
||||
if (existingPromise) {
|
||||
return existingPromise as Promise<Tasks[K]['output']>;
|
||||
}
|
||||
}
|
||||
|
||||
return new Promise<Tasks[K]['output']>((resolve, reject) => {
|
||||
// Supersede any pending task with the same key (may have different name)
|
||||
for (const existingTask of Object.values(this.tasks) as Task[]) {
|
||||
if (existingTask?.key === key && existingTask.status === 'pending') {
|
||||
existingTask.abort.abort(new StoreError('SUPERSEDED'));
|
||||
}
|
||||
}
|
||||
|
||||
const promise = new Promise<Tasks[K]['output']>((resolve, reject) => {
|
||||
this.#executeNow({
|
||||
id: Symbol('@videojs/task'),
|
||||
name,
|
||||
@@ -95,6 +106,18 @@ export class Queue<Tasks extends TaskRecord = DefaultTaskRecord> {
|
||||
reject,
|
||||
});
|
||||
});
|
||||
|
||||
// Track promise for shared mode
|
||||
if (mode === 'shared') {
|
||||
this.#sharedPromises.set(key, promise);
|
||||
// Use .then() to avoid unhandled rejection from .finally() propagating errors
|
||||
promise.then(
|
||||
() => this.#sharedPromises.delete(key),
|
||||
() => this.#sharedPromises.delete(key),
|
||||
);
|
||||
}
|
||||
|
||||
return promise;
|
||||
}
|
||||
|
||||
/** Abort task(s). If name provided, aborts that task. If no name, aborts all. */
|
||||
@@ -127,6 +150,8 @@ export class Queue<Tasks extends TaskRecord = DefaultTaskRecord> {
|
||||
for (const key of Reflect.ownKeys(this.tasks) as (keyof Tasks)[]) {
|
||||
delete this.tasks[key];
|
||||
}
|
||||
|
||||
this.#sharedPromises.clear();
|
||||
}
|
||||
|
||||
async #executeNow<K extends keyof Tasks>(params: {
|
||||
|
||||
@@ -8,7 +8,10 @@ import { isFunction, isObject } from '@videojs/utils/predicate';
|
||||
// Symbols
|
||||
// ----------------------------------------
|
||||
|
||||
export const REQUEST_META: unique symbol = Symbol.for('@videojs/request');
|
||||
export const REQUEST_META = Symbol.for('@videojs/request');
|
||||
|
||||
/** Cancel all pending requests when this request is enqueued. Like native `load()`. */
|
||||
export const CANCEL_ALL = Symbol.for('@videojs/cancel-all');
|
||||
|
||||
// ----------------------------------------
|
||||
// Types
|
||||
@@ -33,7 +36,13 @@ export interface RequestContext<Target> {
|
||||
|
||||
export type RequestKey<Input = unknown> = TaskKey | ((input: Input) => TaskKey);
|
||||
|
||||
export type RequestCancel<Input = unknown> = TaskKey | TaskKey[] | ((input: Input) => TaskKey | TaskKey[]);
|
||||
export type RequestMode = 'exclusive' | 'shared';
|
||||
|
||||
export type RequestCancel<Input = unknown>
|
||||
= | typeof CANCEL_ALL
|
||||
| TaskKey
|
||||
| TaskKey[]
|
||||
| ((input: Input) => typeof CANCEL_ALL | TaskKey | TaskKey[]);
|
||||
|
||||
export type RequestHandler<Target, Input = unknown, Output = unknown> = (
|
||||
input: Input,
|
||||
@@ -42,6 +51,7 @@ export type RequestHandler<Target, Input = unknown, Output = unknown> = (
|
||||
|
||||
export interface RequestConfig<Target, Input = unknown, Output = unknown> {
|
||||
key?: RequestKey<Input>;
|
||||
mode?: RequestMode;
|
||||
guard?: Guard<Target> | Guard<Target>[];
|
||||
cancel?: RequestCancel<Input>;
|
||||
handler: RequestHandler<Target, Input, Output>;
|
||||
@@ -49,6 +59,7 @@ export interface RequestConfig<Target, Input = unknown, Output = unknown> {
|
||||
|
||||
export interface ResolvedRequestConfig<Target, Input = unknown, Output = unknown> {
|
||||
key: RequestKey<Input>;
|
||||
mode: RequestMode;
|
||||
guard: Guard<Target>[];
|
||||
cancel?: RequestCancel<Input> | undefined;
|
||||
handler: RequestHandler<Target, Input, Output>;
|
||||
@@ -112,12 +123,14 @@ export function resolveRequests<Target, Requests extends { [K in keyof Requests]
|
||||
if (isFunction(config)) {
|
||||
resolved[name] = {
|
||||
key: name,
|
||||
mode: 'exclusive',
|
||||
guard: [],
|
||||
handler: config,
|
||||
};
|
||||
} else {
|
||||
resolved[name] = {
|
||||
key: config.key ?? name,
|
||||
mode: config.mode ?? 'exclusive',
|
||||
cancel: config.cancel,
|
||||
handler: config.handler,
|
||||
guard: config.guard ? (Array.isArray(config.guard) ? config.guard : [config.guard]) : [],
|
||||
@@ -132,9 +145,13 @@ export function resolveRequestKey(keyConfig: RequestKey<any>, input: unknown): T
|
||||
return isFunction(keyConfig) ? keyConfig(input) : keyConfig;
|
||||
}
|
||||
|
||||
export function resolveRequestCancel(cancel: RequestCancel<any> | undefined, input: unknown): TaskKey[] {
|
||||
export function resolveRequestCancel(
|
||||
cancel: RequestCancel<any> | undefined,
|
||||
input: unknown,
|
||||
): typeof CANCEL_ALL | TaskKey[] {
|
||||
if (!cancel) return [];
|
||||
const result = isFunction(cancel) ? cancel(input) : cancel;
|
||||
if (result === CANCEL_ALL) return CANCEL_ALL;
|
||||
return Array.isArray(result) ? result : [result];
|
||||
}
|
||||
|
||||
|
||||
@@ -14,7 +14,7 @@ import { isNull } from '@videojs/utils/predicate';
|
||||
|
||||
import { StoreError } from './errors';
|
||||
import { Queue } from './queue';
|
||||
import { createRequestMeta, resolveRequestCancel, resolveRequestKey } from './request';
|
||||
import { CANCEL_ALL, createRequestMeta, resolveRequestCancel, resolveRequestKey } from './request';
|
||||
import { reactive } from './state';
|
||||
|
||||
export class Store<Target, Slices extends AnySlice<Target>[] = AnySlice<Target>[]> {
|
||||
@@ -225,8 +225,13 @@ export class Store<Target, Slices extends AnySlice<Target>[] = AnySlice<Target>[
|
||||
): Promise<unknown> {
|
||||
const key = resolveRequestKey(config.key, input);
|
||||
|
||||
for (const requestName of resolveRequestCancel(config.cancel, input)) {
|
||||
this.#queue.abort(requestName);
|
||||
const cancel = resolveRequestCancel(config.cancel, input);
|
||||
if (cancel === CANCEL_ALL) {
|
||||
this.#queue.abort();
|
||||
} else {
|
||||
for (const requestName of cancel) {
|
||||
this.#queue.abort(requestName);
|
||||
}
|
||||
}
|
||||
|
||||
const handler = async ({ input, signal }: TaskContext) => {
|
||||
@@ -255,6 +260,7 @@ export class Store<Target, Slices extends AnySlice<Target>[] = AnySlice<Target>[
|
||||
return await this.#queue.enqueue({
|
||||
name,
|
||||
key,
|
||||
mode: config.mode,
|
||||
input,
|
||||
meta,
|
||||
handler,
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import { describe, expect, it } from 'vitest';
|
||||
|
||||
import { createSlice, createStore } from '../../index';
|
||||
import { CANCEL_ALL, createSlice, createStore } from '../../index';
|
||||
import { flush, subscribeKeys } from '../../state';
|
||||
|
||||
describe('store lifecycle integration', () => {
|
||||
@@ -232,6 +232,198 @@ describe('request coordination', () => {
|
||||
expect(executed).not.toContain('first-end');
|
||||
expect(executed).not.toContain('second-end');
|
||||
});
|
||||
|
||||
it('CANCEL_ALL aborts all pending requests (nuclear reset)', async () => {
|
||||
const events: string[] = [];
|
||||
|
||||
const slice = createSlice<unknown>()({
|
||||
initialState: {},
|
||||
getSnapshot: () => ({}),
|
||||
subscribe: () => {},
|
||||
request: {
|
||||
fetch: {
|
||||
key: 'fetch',
|
||||
async handler(_, { signal }) {
|
||||
events.push('fetch-start');
|
||||
await new Promise((resolve, reject) => {
|
||||
const timeout = setTimeout(() => {
|
||||
events.push('fetch-complete');
|
||||
resolve('fetched');
|
||||
}, 100);
|
||||
signal.addEventListener('abort', () => {
|
||||
clearTimeout(timeout);
|
||||
events.push('fetch-aborted');
|
||||
reject(signal.reason);
|
||||
});
|
||||
});
|
||||
},
|
||||
},
|
||||
seek: {
|
||||
key: 'seek',
|
||||
async handler(_, { signal }) {
|
||||
events.push('seek-start');
|
||||
await new Promise((resolve, reject) => {
|
||||
const timeout = setTimeout(() => {
|
||||
events.push('seek-complete');
|
||||
resolve('seeked');
|
||||
}, 100);
|
||||
signal.addEventListener('abort', () => {
|
||||
clearTimeout(timeout);
|
||||
events.push('seek-aborted');
|
||||
reject(signal.reason);
|
||||
});
|
||||
});
|
||||
},
|
||||
},
|
||||
load: {
|
||||
cancel: CANCEL_ALL,
|
||||
handler: () => {
|
||||
events.push('load');
|
||||
},
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
const store = createStore({
|
||||
slices: [slice],
|
||||
onError: () => {},
|
||||
});
|
||||
|
||||
store.attach({});
|
||||
|
||||
// Start multiple requests
|
||||
const fetchPromise = store.request.fetch();
|
||||
const seekPromise = store.request.seek();
|
||||
await new Promise(r => setTimeout(r, 10));
|
||||
|
||||
// Nuclear reset
|
||||
await store.request.load();
|
||||
|
||||
await fetchPromise.catch(() => {});
|
||||
await seekPromise.catch(() => {});
|
||||
|
||||
expect(events).toContain('fetch-start');
|
||||
expect(events).toContain('seek-start');
|
||||
expect(events).toContain('fetch-aborted');
|
||||
expect(events).toContain('seek-aborted');
|
||||
expect(events).toContain('load');
|
||||
expect(events).not.toContain('fetch-complete');
|
||||
expect(events).not.toContain('seek-complete');
|
||||
});
|
||||
|
||||
it('mode: shared allows multiple requests to share fate', async () => {
|
||||
let handlerCallCount = 0;
|
||||
|
||||
const slice = createSlice<unknown>()({
|
||||
initialState: {},
|
||||
getSnapshot: () => ({}),
|
||||
subscribe: () => {},
|
||||
request: {
|
||||
play: {
|
||||
key: 'playback',
|
||||
mode: 'shared',
|
||||
async handler() {
|
||||
handlerCallCount++;
|
||||
await new Promise(r => setTimeout(r, 50));
|
||||
return 'playing';
|
||||
},
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
const store = createStore({
|
||||
slices: [slice],
|
||||
});
|
||||
|
||||
store.attach({});
|
||||
|
||||
// Multiple play calls while one is pending
|
||||
const p1 = store.request.play();
|
||||
const p2 = store.request.play();
|
||||
const p3 = store.request.play();
|
||||
|
||||
// All should resolve to the same value (shared fate)
|
||||
const [r1, r2, r3] = await Promise.all([p1, p2, p3]);
|
||||
expect(r1).toBe('playing');
|
||||
expect(r2).toBe('playing');
|
||||
expect(r3).toBe('playing');
|
||||
|
||||
// Handler should only be called once (not superseded)
|
||||
expect(handlerCallCount).toBe(1);
|
||||
});
|
||||
|
||||
it('mode: shared rejects all promises together on error', async () => {
|
||||
const slice = createSlice<unknown>()({
|
||||
initialState: {},
|
||||
getSnapshot: () => ({}),
|
||||
subscribe: () => {},
|
||||
request: {
|
||||
play: {
|
||||
key: 'playback',
|
||||
mode: 'shared',
|
||||
async handler() {
|
||||
await new Promise(r => setTimeout(r, 20));
|
||||
throw new Error('playback failed');
|
||||
},
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
const store = createStore({
|
||||
slices: [slice],
|
||||
onError: () => {},
|
||||
});
|
||||
|
||||
store.attach({});
|
||||
|
||||
const p1 = store.request.play();
|
||||
const p2 = store.request.play();
|
||||
|
||||
// Both should reject with the same error
|
||||
await expect(p1).rejects.toThrow('playback failed');
|
||||
await expect(p2).rejects.toThrow('playback failed');
|
||||
});
|
||||
|
||||
it('mode: shared allows new request after previous completes', async () => {
|
||||
let callCount = 0;
|
||||
|
||||
const slice = createSlice<unknown>()({
|
||||
initialState: {},
|
||||
getSnapshot: () => ({}),
|
||||
subscribe: () => {},
|
||||
request: {
|
||||
play: {
|
||||
key: 'playback',
|
||||
mode: 'shared',
|
||||
async handler() {
|
||||
callCount++;
|
||||
await new Promise(r => setTimeout(r, 10));
|
||||
return `call-${callCount}`;
|
||||
},
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
const store = createStore({
|
||||
slices: [slice],
|
||||
});
|
||||
|
||||
store.attach({});
|
||||
|
||||
// First batch shares fate
|
||||
const p1 = store.request.play();
|
||||
const p2 = store.request.play();
|
||||
const [r1, r2] = await Promise.all([p1, p2]);
|
||||
expect(r1).toBe('call-1');
|
||||
expect(r2).toBe('call-1');
|
||||
|
||||
// After completion, new request starts fresh
|
||||
const p3 = store.request.play();
|
||||
const r3 = await p3;
|
||||
expect(r3).toBe('call-2');
|
||||
|
||||
expect(callCount).toBe(2);
|
||||
});
|
||||
});
|
||||
|
||||
describe('state syncing', () => {
|
||||
|
||||
Reference in New Issue
Block a user