mirror of
https://github.com/vector-im/element-call.git
synced 2026-10-16 23:45:50 +00:00
7000 loc COPIED from element call
They are copied for now but will be removed from element call once it adopts the sdk. Those are no real line additions.
This commit is contained in:
Timo K.
@@ -0,0 +1,29 @@
|
||||
/*
|
||||
Copyright 2025 New Vector Ltd.
|
||||
|
||||
SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
|
||||
Please see LICENSE in the repository root for full details.
|
||||
*/
|
||||
|
||||
import { BehaviorSubject } from "rxjs";
|
||||
|
||||
/**
|
||||
* A stateful, read-only reactive value. As an Observable, it is "hot" and
|
||||
* always replays the current value upon subscription.
|
||||
*
|
||||
* A Behavior is to BehaviorSubject what Observable is to Subject; it does not
|
||||
* provide a way to imperatively set new values. For more info on the
|
||||
* distinction between Behaviors and Observables, see
|
||||
* https://monoid.dk/post/behaviors-and-streams-why-both/.
|
||||
*/
|
||||
export type Behavior<T> = Omit<
|
||||
BehaviorSubject<T>,
|
||||
"next" | "observers" | "error"
|
||||
>;
|
||||
|
||||
/**
|
||||
* Creates a Behavior which never changes in value.
|
||||
*/
|
||||
export function constant<T>(value: T): Behavior<T> {
|
||||
return new BehaviorSubject(value);
|
||||
}
|
||||
@@ -0,0 +1,293 @@
|
||||
/*
|
||||
Copyright 2025 Element Creations Ltd.
|
||||
|
||||
SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
|
||||
Please see LICENSE in the repository root for full details.
|
||||
*/
|
||||
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { BehaviorSubject, combineLatest, Subject } from "rxjs";
|
||||
import { logger } from "matrix-js-sdk/lib/logger";
|
||||
import { sleep } from "matrix-js-sdk/lib/utils";
|
||||
|
||||
import {
|
||||
Epoch,
|
||||
mapEpoch,
|
||||
ObservableScope,
|
||||
trackEpoch,
|
||||
} from "./ObservableScope";
|
||||
import { testScope, withTestScheduler } from "../utils/test";
|
||||
|
||||
describe("Epoch", () => {
|
||||
it("should map the value correctly", () => {
|
||||
const epoch = new Epoch(1);
|
||||
const mappedEpoch = epoch.mapInner((v) => v + 1);
|
||||
expect(mappedEpoch.value).toBe(2);
|
||||
expect(mappedEpoch.epoch).toBe(0);
|
||||
});
|
||||
|
||||
it("should be tracked from an observable", () => {
|
||||
withTestScheduler(({ expectObservable, behavior }) => {
|
||||
const observable$ = behavior("abc", {
|
||||
a: 1,
|
||||
b: 2,
|
||||
c: 3,
|
||||
});
|
||||
const epochObservable$ = observable$.pipe(trackEpoch());
|
||||
expectObservable(epochObservable$).toBe("abc", {
|
||||
a: expect.toSatisfy((e) => e.epoch === 0 && e.value === 1),
|
||||
b: expect.toSatisfy((e) => e.epoch === 1 && e.value === 2),
|
||||
c: expect.toSatisfy((e) => e.epoch === 2 && e.value === 3),
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
it("can be mapped without loosing epoch information", () => {
|
||||
withTestScheduler(({ expectObservable, behavior }) => {
|
||||
const observable$ = behavior("abc", {
|
||||
a: "A",
|
||||
b: "B",
|
||||
c: "C",
|
||||
});
|
||||
const epochObservable$ = observable$.pipe(trackEpoch());
|
||||
const derivedEpoch$ = epochObservable$.pipe(
|
||||
mapEpoch((e) => e + "-mapped"),
|
||||
);
|
||||
|
||||
expectObservable(derivedEpoch$).toBe("abc", {
|
||||
a: new Epoch("A-mapped", 0),
|
||||
b: new Epoch("B-mapped", 1),
|
||||
c: new Epoch("C-mapped", 2),
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
it("diamonds emits in a predictable order", () => {
|
||||
const sb$ = new BehaviorSubject("initial");
|
||||
const root$ = sb$.pipe(trackEpoch());
|
||||
const derivedA$ = root$.pipe(mapEpoch((e) => e + "-A"));
|
||||
const derivedB$ = root$.pipe(mapEpoch((e) => e + "-B"));
|
||||
combineLatest([root$, derivedB$, derivedA$]).subscribe(
|
||||
([root, derivedA, derivedB]) => {
|
||||
logger.log(
|
||||
"combined" +
|
||||
root.epoch +
|
||||
root.value +
|
||||
"\n" +
|
||||
derivedA.epoch +
|
||||
derivedA.value +
|
||||
"\n" +
|
||||
derivedB.epoch +
|
||||
derivedB.value,
|
||||
);
|
||||
},
|
||||
);
|
||||
sb$.next("updated");
|
||||
sb$.next("ANOTERUPDATE");
|
||||
});
|
||||
|
||||
it("behavior test", () => {
|
||||
const scope = testScope();
|
||||
const s$ = new Subject();
|
||||
const behavior$ = scope.behavior(s$, 0);
|
||||
behavior$.subscribe((value) => {
|
||||
logger.log(`Received value: ${value}`);
|
||||
});
|
||||
s$.next(1);
|
||||
s$.next(2);
|
||||
s$.next(3);
|
||||
s$.next(3);
|
||||
s$.next(3);
|
||||
s$.next(3);
|
||||
s$.next(3);
|
||||
s$.complete();
|
||||
});
|
||||
});
|
||||
|
||||
describe("Reconcile", () => {
|
||||
beforeEach(() => {
|
||||
vi.useFakeTimers();
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
vi.useRealTimers();
|
||||
});
|
||||
|
||||
it("should wait clean up before processing next", async () => {
|
||||
vi.useFakeTimers();
|
||||
const scope = testScope();
|
||||
const behavior$ = new BehaviorSubject<number>(0);
|
||||
|
||||
const setup = vi.fn().mockImplementation(async () => await sleep(100));
|
||||
const cleanup = vi
|
||||
.fn()
|
||||
.mockImplementation(async (n: number) => await sleep(100));
|
||||
scope.reconcile(behavior$, async (value) => {
|
||||
await setup();
|
||||
return async (): Promise<void> => {
|
||||
await cleanup(value);
|
||||
};
|
||||
});
|
||||
// Let the initial setup process
|
||||
await vi.advanceTimersByTimeAsync(120);
|
||||
expect(setup).toHaveBeenCalledTimes(1);
|
||||
expect(cleanup).toHaveBeenCalledTimes(0);
|
||||
|
||||
// Send next value
|
||||
behavior$.next(1);
|
||||
await vi.advanceTimersByTimeAsync(50);
|
||||
// Should not have started setup for 1 yet
|
||||
expect(setup).toHaveBeenCalledTimes(1);
|
||||
expect(cleanup).toHaveBeenCalledTimes(1);
|
||||
expect(cleanup).toHaveBeenCalledWith(0);
|
||||
|
||||
// Let cleanup finish
|
||||
await vi.advanceTimersByTimeAsync(50);
|
||||
// Now setup for 1 should have started
|
||||
expect(setup).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it("should skip intermediates values that are not setup", async () => {
|
||||
vi.useFakeTimers();
|
||||
const scope = testScope();
|
||||
const behavior$ = new BehaviorSubject<number>(0);
|
||||
|
||||
const setup = vi
|
||||
.fn()
|
||||
.mockImplementation(async (n: number) => await sleep(100));
|
||||
|
||||
const cleanupLock = Promise.withResolvers();
|
||||
const cleanup = vi
|
||||
.fn()
|
||||
.mockImplementation(async (n: number) => await cleanupLock.promise);
|
||||
|
||||
scope.reconcile(behavior$, async (value) => {
|
||||
await setup(value);
|
||||
return async (): Promise<void> => {
|
||||
await cleanup(value);
|
||||
};
|
||||
});
|
||||
// Let the initial setup process (0)
|
||||
await vi.advanceTimersByTimeAsync(120);
|
||||
|
||||
// Send 4 next values quickly
|
||||
behavior$.next(1);
|
||||
behavior$.next(2);
|
||||
behavior$.next(3);
|
||||
behavior$.next(4);
|
||||
|
||||
await vi.advanceTimersByTimeAsync(3000);
|
||||
// should have only called cleanup for 0
|
||||
expect(cleanup).toHaveBeenCalledTimes(1);
|
||||
expect(cleanup).toHaveBeenCalledWith(0);
|
||||
// Let cleanup finish
|
||||
cleanupLock.resolve(undefined);
|
||||
await vi.advanceTimersByTimeAsync(120);
|
||||
|
||||
// Now setup for 4 should have started, skipping 1,2,3
|
||||
expect(setup).toHaveBeenCalledTimes(2);
|
||||
expect(setup).toHaveBeenCalledWith(4);
|
||||
expect(setup).not.toHaveBeenCalledWith(1);
|
||||
expect(setup).not.toHaveBeenCalledWith(2);
|
||||
expect(setup).not.toHaveBeenCalledWith(3);
|
||||
});
|
||||
|
||||
it("should wait for setup to complete before starting cleanup", async () => {
|
||||
vi.useFakeTimers();
|
||||
const scope = testScope();
|
||||
const behavior$ = new BehaviorSubject<number>(0);
|
||||
|
||||
const setup = vi
|
||||
.fn()
|
||||
.mockImplementation(async (n: number) => await sleep(3000));
|
||||
|
||||
const cleanupLock = Promise.withResolvers();
|
||||
const cleanup = vi
|
||||
.fn()
|
||||
.mockImplementation(async (n: number) => await cleanupLock.promise);
|
||||
|
||||
scope.reconcile(behavior$, async (value) => {
|
||||
await setup(value);
|
||||
return async (): Promise<void> => {
|
||||
await cleanup(value);
|
||||
};
|
||||
});
|
||||
|
||||
await vi.advanceTimersByTimeAsync(500);
|
||||
// Setup for 0 should be in progress
|
||||
expect(setup).toHaveBeenCalledTimes(1);
|
||||
|
||||
behavior$.next(1);
|
||||
await vi.advanceTimersByTimeAsync(500);
|
||||
|
||||
// Should not have started setup for 1 yet
|
||||
expect(setup).not.toHaveBeenCalledWith(1);
|
||||
// Should not have called cleanup yet, because the setup for 0 is not done
|
||||
expect(cleanup).toHaveBeenCalledTimes(0);
|
||||
|
||||
// Let setup for 0 finish
|
||||
await vi.advanceTimersByTimeAsync(2500 + 100);
|
||||
// Now cleanup for 0 should have started
|
||||
expect(cleanup).toHaveBeenCalledTimes(1);
|
||||
expect(cleanup).toHaveBeenCalledWith(0);
|
||||
|
||||
cleanupLock.resolve(undefined);
|
||||
await vi.advanceTimersByTimeAsync(100);
|
||||
// Now setup for 1 should have started
|
||||
expect(setup).toHaveBeenCalledWith(1);
|
||||
});
|
||||
});
|
||||
|
||||
describe("behavior", () => {
|
||||
it("warns with the tag and nesting depth, logging each new depth once", () => {
|
||||
const warn = vi.spyOn(logger, "warn").mockImplementation(() => {});
|
||||
const scope = testScope();
|
||||
const source$ = new Subject<number>();
|
||||
const behavior$ = scope.behavior(source$, 0, "tagged");
|
||||
// Re-enter once on value 1 (depth 2), then a nested re-entry on value 2
|
||||
// (depth 3), which is the shape needed to strand a middle subscriber.
|
||||
behavior$.subscribe((v) => {
|
||||
if (v === 1) source$.next(2);
|
||||
});
|
||||
behavior$.subscribe((v) => {
|
||||
if (v === 2) source$.next(3);
|
||||
});
|
||||
const seen: number[] = [];
|
||||
behavior$.subscribe((v) => seen.push(v));
|
||||
|
||||
source$.next(1);
|
||||
|
||||
const messages = warn.mock.calls.map((c) => c[0] as string);
|
||||
expect(messages).toEqual([
|
||||
expect.stringContaining("Behavior (tagged) re-entered at depth 2"),
|
||||
expect.stringContaining("Behavior (tagged) re-entered at depth 3"),
|
||||
]);
|
||||
warn.mock.calls.forEach((c) => expect(c[1]).toEqual(expect.any(String)));
|
||||
// The behavior settles on the newest value while the last subscriber is
|
||||
// left stranded on the oldest.
|
||||
expect(behavior$.value).toBe(3);
|
||||
expect(seen.at(-1)).toBe(1);
|
||||
});
|
||||
|
||||
it("warns when the scope is ended while a behavior is mid-delivery", () => {
|
||||
const warn = vi.spyOn(logger, "warn").mockImplementation(() => {});
|
||||
const scope = new ObservableScope();
|
||||
const source$ = new Subject<number>();
|
||||
const behavior$ = scope.behavior(source$, 0);
|
||||
behavior$.subscribe((v) => {
|
||||
if (v === 1) scope.end();
|
||||
});
|
||||
// A derived behavior is bound to the scope, so ending the scope
|
||||
// mid-delivery unsubscribes it before the value reaches it.
|
||||
const derived$ = scope.behavior(behavior$);
|
||||
|
||||
source$.next(1);
|
||||
|
||||
expect(warn).toHaveBeenCalledWith(
|
||||
expect.stringContaining("Scope ended while 1 of its behaviors"),
|
||||
expect.any(String),
|
||||
);
|
||||
expect(behavior$.value).toBe(1);
|
||||
expect(derived$.value).toBe(0);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,329 @@
|
||||
/*
|
||||
Copyright 2024 New Vector Ltd.
|
||||
|
||||
SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
|
||||
Please see LICENSE in the repository root for full details.
|
||||
*/
|
||||
|
||||
import {
|
||||
BehaviorSubject,
|
||||
catchError,
|
||||
distinctUntilChanged,
|
||||
EMPTY,
|
||||
endWith,
|
||||
filter,
|
||||
map,
|
||||
type Observable,
|
||||
type OperatorFunction,
|
||||
share,
|
||||
take,
|
||||
takeUntil,
|
||||
} from "rxjs";
|
||||
|
||||
import { logger } from "matrix-js-sdk/lib/logger";
|
||||
|
||||
import { type Behavior } from "./Behavior";
|
||||
|
||||
type MonoTypeOperator = <T>(o: Observable<T>) => Observable<T>;
|
||||
|
||||
type SplitBehavior<T> = keyof T extends string | number
|
||||
? { [K in keyof T as `${K}$`]: Behavior<T[K]> }
|
||||
: never;
|
||||
|
||||
const nothing = Symbol("nothing");
|
||||
|
||||
/**
|
||||
* A scope which limits the execution lifetime of its bound Observables.
|
||||
*/
|
||||
export class ObservableScope {
|
||||
/**
|
||||
* Number of this scope's behaviors currently mid-delivery. Used to detect a
|
||||
* scope being torn down synchronously from within one of its own emissions.
|
||||
*/
|
||||
private delivering = 0;
|
||||
|
||||
private readonly ended$ = new BehaviorSubject(false);
|
||||
|
||||
private readonly bindImpl: MonoTypeOperator = takeUntil(
|
||||
this.ended$.pipe(filter((ended) => ended)),
|
||||
);
|
||||
|
||||
/**
|
||||
* Binds an Observable to this scope, so that it completes when the scope
|
||||
* ends.
|
||||
*/
|
||||
public bind(): MonoTypeOperator {
|
||||
return this.bindImpl;
|
||||
}
|
||||
|
||||
private readonly shareImpl: MonoTypeOperator = share({
|
||||
resetOnError: false,
|
||||
resetOnComplete: false,
|
||||
resetOnRefCountZero: false,
|
||||
});
|
||||
/**
|
||||
* Shares (multicasts) the Observable as a hot Observable.
|
||||
*/
|
||||
public readonly share: MonoTypeOperator = (input$) =>
|
||||
input$.pipe(this.bindImpl, this.shareImpl);
|
||||
|
||||
/**
|
||||
* Converts an Observable to a Behavior. If no initial value is specified, the
|
||||
* Observable must synchronously emit an initial value.
|
||||
*/
|
||||
public behavior<T>(
|
||||
setValue$: Observable<T>,
|
||||
initialValue: T | typeof nothing = nothing,
|
||||
// An optional label used only in the re-entrancy warning below, so that a
|
||||
// torn behavior can be identified from a rageshake.
|
||||
tag?: string,
|
||||
): Behavior<T> {
|
||||
const subject$ = new BehaviorSubject(initialValue);
|
||||
// Push values from the Observable into the BehaviorSubject.
|
||||
// BehaviorSubjects have an undesirable feature where if you call 'complete',
|
||||
// they will no longer re-emit their current value upon subscription. We want
|
||||
// to support Observables that complete (for example `of({})`), so we have to
|
||||
// take care to not propagate the completion event.
|
||||
// If a subscriber synchronously causes this same behavior to emit again,
|
||||
// rxjs delivers the nested value to every subscriber first and then
|
||||
// resumes delivering the outer (older) value to the remaining subscribers,
|
||||
// leaving them permanently out of sync with the others. A single such
|
||||
// re-entry can only strand a contiguous run of subscribers; stranding a
|
||||
// subscriber in the middle of the list needs a nested (deeper) re-entry, so
|
||||
// we report the depth and log each new depth (with a stack trace) rather
|
||||
// than only the first occurrence, to make a nested re-entry visible.
|
||||
let depth = 0;
|
||||
let maxReportedDepth = 1;
|
||||
setValue$.pipe(this.bind(), distinctUntilChanged()).subscribe({
|
||||
next: (value) => {
|
||||
if (depth > 0 && depth + 1 > maxReportedDepth) {
|
||||
maxReportedDepth = depth + 1;
|
||||
logger.warn(
|
||||
`Behavior${tag ? ` (${tag})` : ""} re-entered at depth ${
|
||||
depth + 1
|
||||
} while delivering a value; later subscribers will be left with a stale value`,
|
||||
new Error().stack,
|
||||
);
|
||||
}
|
||||
depth++;
|
||||
this.delivering++;
|
||||
try {
|
||||
subject$.next(value);
|
||||
} finally {
|
||||
depth--;
|
||||
this.delivering--;
|
||||
}
|
||||
},
|
||||
error: (err: unknown) => {
|
||||
subject$.error(err);
|
||||
},
|
||||
});
|
||||
if (subject$.value === nothing)
|
||||
throw new Error("Behavior failed to synchronously emit an initial value");
|
||||
return subject$ as Behavior<T>;
|
||||
}
|
||||
|
||||
/**
|
||||
* Ends the scope, causing any bound Observables to complete.
|
||||
*/
|
||||
public end(): void {
|
||||
if (this.delivering > 0)
|
||||
logger.warn(
|
||||
`Scope ended while ${this.delivering} of its behaviors were still delivering a value; later subscribers will be left with a stale value`,
|
||||
new Error().stack,
|
||||
);
|
||||
this.ended$.next(true);
|
||||
}
|
||||
|
||||
/**
|
||||
* Register a callback to be executed when the scope is ended.
|
||||
*/
|
||||
public onEnd(callback: () => void): void {
|
||||
this.ended$
|
||||
.pipe(
|
||||
filter((ended) => ended),
|
||||
take(1),
|
||||
)
|
||||
.subscribe(callback);
|
||||
}
|
||||
|
||||
/**
|
||||
* For the duration of the scope, sync some external state with the value of
|
||||
* the provided Behavior by way of an async function which attempts to update
|
||||
* (reconcile) the external state. The reconciliation function may return a
|
||||
* clean-up callback which will be called and awaited before the next change
|
||||
* in value (or the end of the scope).
|
||||
*
|
||||
* All calls to the function and its clean-up callbacks are serialized. If the
|
||||
* value changes faster than the handlers can keep up with, intermediate
|
||||
* values may be skipped.
|
||||
*
|
||||
* Basically, this is like React's useEffect but async and for Behaviors.
|
||||
*/
|
||||
public reconcile<T>(
|
||||
value$: Behavior<T>,
|
||||
callback: (value: T) => Promise<(() => Promise<void>) | void>,
|
||||
): void {
|
||||
let latestValue: T | typeof nothing = nothing;
|
||||
let reconcilePromise: Promise<void> | undefined = undefined;
|
||||
let cleanUp: (() => Promise<void>) | void = undefined;
|
||||
let prevVal: T | typeof nothing = nothing;
|
||||
|
||||
// While this loop runs it will process the latest from `value$` until it caught up with the updates.
|
||||
// It might skip updates from `value$` and only process the newest value after callback has resolved.
|
||||
const reconcileLoop = async (): Promise<void> => {
|
||||
while (latestValue !== prevVal) {
|
||||
await cleanUp?.(); // Call the previous value's clean-up handler
|
||||
prevVal = latestValue;
|
||||
|
||||
if (latestValue !== nothing) cleanUp = await callback(latestValue); // Sync current value...
|
||||
// `latestValue` might have gotten updated during the `await callback`. That is why we loop here
|
||||
}
|
||||
};
|
||||
|
||||
value$
|
||||
.pipe(
|
||||
catchError(() => EMPTY), // Ignore errors
|
||||
this.bind(), // Limit to the duration of the scope
|
||||
endWith(nothing), // Clean up when the scope ends
|
||||
)
|
||||
.subscribe((value) => {
|
||||
// Always track the latest value! The `reconcileLoop` will run until it "processed" the "last" `latestValue`.
|
||||
latestValue = value;
|
||||
// There's already an instance of the below 'reconcileLoop' loop running
|
||||
// concurrently. So lets let the loop handle it. NEVER instanciate two `reconcileLoop`s.
|
||||
if (reconcilePromise) return;
|
||||
|
||||
reconcilePromise = reconcileLoop().finally(() => {
|
||||
reconcilePromise = undefined;
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Splits a Behavior of objects with static properties into an object with
|
||||
* Behavior properties.
|
||||
*
|
||||
* For example, splitting a `Behavior<{ name: string; age: number }>` results
|
||||
* in an object of type `{ name$: Behavior<string>; age$: Behavior<number> }`.
|
||||
*/
|
||||
public splitBehavior<T extends object>(
|
||||
input$: Behavior<T>,
|
||||
): SplitBehavior<T> {
|
||||
return Object.fromEntries(
|
||||
Object.keys(input$.value).map((key) => [
|
||||
`${key}$`,
|
||||
this.behavior(
|
||||
input$.pipe(map((input) => input[key as keyof T])),
|
||||
nothing,
|
||||
key,
|
||||
),
|
||||
]),
|
||||
) as SplitBehavior<T>;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* `Epoch`'s can be used to create `Behavior`s and `Observable`s which derivitives can be merged
|
||||
* with `combinedLatest` without duplicated emissions.
|
||||
*
|
||||
* This is useful in the following example:
|
||||
* ```
|
||||
* const rootObs$ = of("red","green","blue");
|
||||
* const derivedObs$ = rootObs$.pipe(
|
||||
* map((v)=> {red:"fire", green:"grass", blue:"water"}[v])
|
||||
* );
|
||||
* const otherDerivedObs$ = rootObs$.pipe(
|
||||
* map((v)=> {red:"tomatoes", green:"leaves", blue:"sky"}[v])
|
||||
* );
|
||||
* const mergedObs$ = combineLatest([rootObs$, derivedObs$, otherDerivedObs$]).pipe(
|
||||
* map(([color, a,b]) => color + " like " + a + " and " + b)
|
||||
* );
|
||||
*
|
||||
* ```
|
||||
* will result in 6 emissions with mismatching items like "red like fire and leaves"
|
||||
*
|
||||
* # Use Epoch
|
||||
* ```
|
||||
* const ancestorObs$ = of(1,2,3).pipe(trackEpoch());
|
||||
* const derivedObs$ = ancestorObs$.pipe(
|
||||
* mapEpoch((v)=> "this number: " + v)
|
||||
* );
|
||||
* const otherDerivedObs$ = ancestorObs$.pipe(
|
||||
* mapEpoch((v)=> "multiplied by: " + v)
|
||||
* );
|
||||
* const mergedObs$ = combineLatest([derivedObs$, otherDerivedObs$]).pipe(
|
||||
* filter((values) => values.every((v) => v.epoch === values[0].v)),
|
||||
* map(([color, a, b]) => color + " like " + a + " and " + b)
|
||||
* );
|
||||
*
|
||||
* ```
|
||||
* will result in 3 emissions all matching (e.g. "blue like water and sky")
|
||||
*/
|
||||
export class Epoch<T> {
|
||||
public readonly epoch: number;
|
||||
public readonly value: T;
|
||||
|
||||
public constructor(value: T, epoch?: number) {
|
||||
this.value = value;
|
||||
this.epoch = epoch ?? 0;
|
||||
}
|
||||
/**
|
||||
* Maps the value inside the epoch to a new value while keeping the epoch number.
|
||||
* # usage
|
||||
* ```
|
||||
* const myEpoch$ = myObservable$.pipe(
|
||||
* map(trackEpoch()),
|
||||
* // this is the preferred way using mapEpoch
|
||||
* mapEpoch((v)=> v+1)
|
||||
* // This is how inner map can be used:
|
||||
* map((epoch) => epoch.innerMap((v)=> v+1))
|
||||
* // It is equivalent to:
|
||||
* map((epoch) => new Epoch(epoch.value + 1, epoch.epoch))
|
||||
* )
|
||||
* ```
|
||||
* See also `Epoch<T>`
|
||||
*/
|
||||
public mapInner<U>(map: (value: T) => U): Epoch<U> {
|
||||
return new Epoch<U>(map(this.value), this.epoch);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* A `pipe` compatible map oparator that keeps the epoch in tact but allows mapping the value.
|
||||
* # usage
|
||||
* ```
|
||||
* const myEpoch$ = myObservable$.pipe(
|
||||
* map(trackEpoch()),
|
||||
* // this is the preferred way using mapEpoch
|
||||
* mapEpoch((v)=> v+1)
|
||||
* // This is how inner map can be used:
|
||||
* map((epoch) => epoch.innerMap((v)=> v+1))
|
||||
* // It is equivalent to:
|
||||
* map((epoch) => new Epoch(epoch.value + 1, epoch.epoch))
|
||||
* )
|
||||
* ```
|
||||
* See also `Epoch<T>`
|
||||
*/
|
||||
export function mapEpoch<T, U>(
|
||||
mapFn: (value: T) => U,
|
||||
): OperatorFunction<Epoch<T>, Epoch<U>> {
|
||||
return map((e) => e.mapInner(mapFn));
|
||||
}
|
||||
|
||||
/**
|
||||
* # usage
|
||||
* ```
|
||||
* const myEpoch$ = myObservable$.pipe(
|
||||
* map(trackEpoch()),
|
||||
* map((epoch) => epoch.innerMap((v)=> v+1))
|
||||
* )
|
||||
* const derived = myEpoch$.pipe(
|
||||
* mapEpoch((v)=>v^2)
|
||||
* )
|
||||
* ```
|
||||
* See also `Epoch<T>`
|
||||
*/
|
||||
export function trackEpoch<T>(): OperatorFunction<T, Epoch<T>> {
|
||||
return map<T, Epoch<T>>((value, number) => new Epoch(value, number));
|
||||
}
|
||||
@@ -0,0 +1,104 @@
|
||||
/*
|
||||
Copyright 2025 New Vector Ltd.
|
||||
|
||||
SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
|
||||
Please see LICENSE in the repository root for full details.
|
||||
*/
|
||||
|
||||
import { expect, test } from "vitest";
|
||||
import { type Observable, of, Subject, switchMap } from "rxjs";
|
||||
|
||||
import { withTestScheduler } from "../utils/test";
|
||||
import { filterBehavior, generateItems, pauseWhen } from "./observable";
|
||||
import { type Behavior } from "./Behavior";
|
||||
|
||||
test("pauseWhen", () => {
|
||||
withTestScheduler(({ behavior, expectObservable }) => {
|
||||
const inputMarbles = " abcdefgh-i-jk-";
|
||||
const pauseMarbles = " n-y--n-yn-y--n";
|
||||
const outputMarbles = "abc--fgh-i---k";
|
||||
expectObservable(
|
||||
behavior(inputMarbles).pipe(
|
||||
pauseWhen(behavior(pauseMarbles, { y: true, n: false })),
|
||||
),
|
||||
).toBe(outputMarbles);
|
||||
});
|
||||
});
|
||||
|
||||
test("generateItems", () => {
|
||||
const scope1$ = new Subject<string>();
|
||||
const scope2$ = new Subject<string>();
|
||||
const scope3$ = new Subject<string>();
|
||||
const scope4$ = new Subject<string>();
|
||||
const scopeSubjects = [scope1$, scope2$, scope3$, scope4$];
|
||||
|
||||
withTestScheduler(({ hot, expectObservable }) => {
|
||||
// Each scope should start when the input number reaches or surpasses their
|
||||
// number and end when the input number drops back below their number.
|
||||
// At the very end, unsubscribing should end all remaining scopes.
|
||||
const inputMarbles = " 123242";
|
||||
const outputMarbles = " abcbdb";
|
||||
const subscriptionMarbles = "^-----!";
|
||||
const scope1Marbles = " y-----n";
|
||||
const scope2Marbles = " -y----n";
|
||||
const scope3Marbles = " --ynyn";
|
||||
const scope4Marbles = " ----yn";
|
||||
|
||||
expectObservable(
|
||||
hot<string>(inputMarbles).pipe(
|
||||
generateItems(
|
||||
"test items",
|
||||
function* (input) {
|
||||
for (let i = 1; i <= +input; i++) {
|
||||
yield { keys: [i], data: undefined };
|
||||
}
|
||||
},
|
||||
(scope, data$, i) => {
|
||||
scopeSubjects[i - 1].next("y");
|
||||
scope.onEnd(() => scopeSubjects[i - 1].next("n"));
|
||||
return i.toString();
|
||||
},
|
||||
),
|
||||
),
|
||||
subscriptionMarbles,
|
||||
).toBe(outputMarbles, {
|
||||
a: ["1"],
|
||||
b: ["1", "2"],
|
||||
c: ["1", "2", "3"],
|
||||
d: ["1", "2", "3", "4"],
|
||||
});
|
||||
|
||||
expectObservable(scope1$).toBe(scope1Marbles);
|
||||
expectObservable(scope2$).toBe(scope2Marbles);
|
||||
expectObservable(scope3$).toBe(scope3Marbles);
|
||||
expectObservable(scope4$).toBe(scope4Marbles);
|
||||
});
|
||||
});
|
||||
|
||||
test("filterBehavior", () => {
|
||||
withTestScheduler(({ behavior, expectObservable }) => {
|
||||
// Filtering the input should segment it into 2 modes of non-null behavior.
|
||||
const inputMarbles = " abcxabx";
|
||||
const filteredMarbles = "a--xa-x";
|
||||
|
||||
const input$ = behavior(inputMarbles, {
|
||||
a: "a",
|
||||
b: "b",
|
||||
c: "c",
|
||||
x: null,
|
||||
});
|
||||
const filtered$: Observable<Behavior<string> | null> = input$.pipe(
|
||||
filterBehavior((value) => typeof value === "string"),
|
||||
);
|
||||
|
||||
expectObservable(filtered$).toBe(filteredMarbles, {
|
||||
a: expect.any(Object),
|
||||
x: null,
|
||||
});
|
||||
expectObservable(
|
||||
filtered$.pipe(
|
||||
switchMap((value$) => (value$ === null ? of(null) : value$)),
|
||||
),
|
||||
).toBe(inputMarbles, { a: "a", b: "b", c: "c", x: null });
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,264 @@
|
||||
/*
|
||||
Copyright 2023, 2024 New Vector Ltd.
|
||||
|
||||
SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
|
||||
Please see LICENSE in the repository root for full details.
|
||||
*/
|
||||
|
||||
import {
|
||||
type Observable,
|
||||
audit,
|
||||
defer,
|
||||
filter,
|
||||
finalize,
|
||||
map,
|
||||
of,
|
||||
scan,
|
||||
tap,
|
||||
withLatestFrom,
|
||||
BehaviorSubject,
|
||||
type OperatorFunction,
|
||||
distinctUntilChanged,
|
||||
} from "rxjs";
|
||||
import { logger } from "matrix-js-sdk/lib/logger";
|
||||
|
||||
import { type Behavior } from "./Behavior";
|
||||
import { Epoch, ObservableScope } from "./ObservableScope";
|
||||
|
||||
const nothing = Symbol("nothing");
|
||||
|
||||
/**
|
||||
* RxJS operator that invokes a callback when the Observable is finalized,
|
||||
* passing the most recently emitted value. If no value was emitted, the
|
||||
* callback will not be invoked.
|
||||
*/
|
||||
export function finalizeValue<T>(callback: (finalValue: T) => void) {
|
||||
return (source$: Observable<T>): Observable<T> =>
|
||||
defer(() => {
|
||||
let finalValue: T | typeof nothing = nothing;
|
||||
return source$.pipe(
|
||||
tap((value) => (finalValue = value)),
|
||||
finalize(() => {
|
||||
if (finalValue !== nothing) callback(finalValue);
|
||||
}),
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* RxJS operator that pauses all changes in the input value whenever a Behavior
|
||||
* is true. When the Behavior returns to being false, the most recently
|
||||
* suppressed change is emitted as the most recent value.
|
||||
*/
|
||||
export function pauseWhen<T>(pause$: Behavior<boolean>) {
|
||||
return (value$: Observable<T>): Observable<T> =>
|
||||
value$.pipe(
|
||||
withLatestFrom(pause$),
|
||||
audit(([, pause]) =>
|
||||
pause ? pause$.pipe(filter((pause) => !pause)) : of(null),
|
||||
),
|
||||
map(([value]) => value),
|
||||
);
|
||||
}
|
||||
|
||||
interface ItemHandle<Keys extends unknown[], Data, Item> {
|
||||
scope: ObservableScope;
|
||||
keys: readonly [...Keys];
|
||||
data$: BehaviorSubject<Data>;
|
||||
item: Item;
|
||||
}
|
||||
|
||||
/**
|
||||
* Maps a changing input value to a collection of items that each capture some
|
||||
* dynamic data and are tied to a key. Items will be automatically created when
|
||||
* their key is requested for the first time, reused when the same key is
|
||||
* requested at a later time, and destroyed (have their scope ended) when the
|
||||
* key is no longer requested.
|
||||
*
|
||||
* @param name A name for this collection to use in debug logs.
|
||||
* @param generator A generator function yielding a tuple of keys and the
|
||||
* currently associated data for each item that it wants to exist.
|
||||
* @param factory A function constructing an individual item, given the item's key,
|
||||
* dynamic data, and an automatically managed ObservableScope for the item.
|
||||
*/
|
||||
export function generateItems<
|
||||
Input,
|
||||
Keys extends [unknown, ...unknown[]],
|
||||
Data,
|
||||
Item,
|
||||
>(
|
||||
name: string,
|
||||
generator: (
|
||||
input: Input,
|
||||
) => Iterable<{ keys: readonly [...Keys]; data: Data }, void, void>,
|
||||
factory: (
|
||||
scope: ObservableScope,
|
||||
data$: Behavior<Data>,
|
||||
...keys: Keys
|
||||
) => Item,
|
||||
): OperatorFunction<Input, Item[]> {
|
||||
return generateItemsInternal(name, generator, factory, (items) => items);
|
||||
}
|
||||
|
||||
/**
|
||||
* Same as generateItems, but preserves epoch data.
|
||||
*/
|
||||
export function generateItemsWithEpoch<
|
||||
Input,
|
||||
Keys extends [unknown, ...unknown[]],
|
||||
Data,
|
||||
Item,
|
||||
>(
|
||||
name: string,
|
||||
generator: (
|
||||
input: Input,
|
||||
) => Iterable<{ keys: readonly [...Keys]; data: Data }, void, void>,
|
||||
factory: (
|
||||
scope: ObservableScope,
|
||||
data$: Behavior<Data>,
|
||||
...keys: Keys
|
||||
) => Item,
|
||||
): OperatorFunction<Epoch<Input>, Epoch<Item[]>> {
|
||||
return generateItemsInternal(
|
||||
name,
|
||||
function* (input) {
|
||||
yield* generator(input.value);
|
||||
},
|
||||
factory,
|
||||
(items, input) => new Epoch(items, input.epoch),
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Segments a behavior into periods during which its value matches the filter
|
||||
* (outputting a behavior with a narrowed type) and periods during which it does
|
||||
* not match (outputting null).
|
||||
*/
|
||||
export function filterBehavior<T, S extends T>(
|
||||
predicate: (value: T) => value is S,
|
||||
): OperatorFunction<T, Behavior<S> | null> {
|
||||
return (input$) =>
|
||||
input$.pipe(
|
||||
scan<T, BehaviorSubject<S> | null>((acc$, input) => {
|
||||
if (predicate(input)) {
|
||||
const output$ = acc$ ?? new BehaviorSubject(input);
|
||||
output$.next(input);
|
||||
return output$;
|
||||
}
|
||||
return null;
|
||||
}, null),
|
||||
distinctUntilChanged(),
|
||||
);
|
||||
}
|
||||
|
||||
function generateItemsInternal<
|
||||
Input,
|
||||
Keys extends [unknown, ...unknown[]],
|
||||
Data,
|
||||
Item,
|
||||
Output,
|
||||
>(
|
||||
name: string,
|
||||
generator: (
|
||||
input: Input,
|
||||
) => Iterable<{ keys: readonly [...Keys]; data: Data }, void, void>,
|
||||
factory: (
|
||||
scope: ObservableScope,
|
||||
data$: Behavior<Data>,
|
||||
...keys: Keys
|
||||
) => Item,
|
||||
project: (items: Item[], input: Input) => Output,
|
||||
): OperatorFunction<Input, Output> {
|
||||
/* eslint-disable @typescript-eslint/no-explicit-any */
|
||||
return (input$) =>
|
||||
input$.pipe(
|
||||
// Keep track of the existing items over time, so they can persist
|
||||
scan<
|
||||
Input,
|
||||
{
|
||||
map: Map<any, any>;
|
||||
items: Set<ItemHandle<Keys, Data, Item>>;
|
||||
input: Input;
|
||||
},
|
||||
{ map: Map<any, any>; items: Set<ItemHandle<Keys, Data, Item>> }
|
||||
>(
|
||||
({ map: prevMap, items: prevItems }, input) => {
|
||||
const nextMap = new Map();
|
||||
const nextItems = new Set<ItemHandle<Keys, Data, Item>>();
|
||||
|
||||
for (const { keys, data } of generator(input)) {
|
||||
// Disable type checks for a second to grab the item out of a nested map
|
||||
let i: any = prevMap;
|
||||
for (const key of keys) i = i?.get(key);
|
||||
let item = i as ItemHandle<Keys, Data, Item> | undefined;
|
||||
|
||||
if (item === undefined) {
|
||||
// First time requesting the key; create the item
|
||||
const scope = new ObservableScope();
|
||||
const data$ = new BehaviorSubject(data);
|
||||
logger.debug(
|
||||
`[${name}] Creating item with keys ${keys.join(", ")}`,
|
||||
);
|
||||
item = {
|
||||
scope,
|
||||
keys,
|
||||
data$,
|
||||
item: factory(scope, data$, ...keys),
|
||||
};
|
||||
} else {
|
||||
item.data$.next(data);
|
||||
}
|
||||
|
||||
// Likewise, disable type checks to insert the item in the nested map
|
||||
let m: Map<any, any> = nextMap;
|
||||
for (let i = 0; i < keys.length - 1; i++) {
|
||||
let inner = m.get(keys[i]);
|
||||
if (inner === undefined) {
|
||||
inner = new Map();
|
||||
m.set(keys[i], inner);
|
||||
}
|
||||
m = inner;
|
||||
}
|
||||
const finalKey = keys[keys.length - 1];
|
||||
if (m.has(finalKey))
|
||||
throw new Error(
|
||||
`Keys must be unique (tried to generate multiple items for key ${keys.join(", ")})`,
|
||||
);
|
||||
m.set(keys[keys.length - 1], item);
|
||||
nextItems.add(item);
|
||||
}
|
||||
|
||||
// Destroy all items that are no longer being requested
|
||||
for (const item of prevItems)
|
||||
if (!nextItems.has(item)) {
|
||||
logger.debug(
|
||||
`[${name}] Destroying item with keys ${item.keys.join(", ")}`,
|
||||
);
|
||||
item.scope.end();
|
||||
}
|
||||
|
||||
return { map: nextMap, items: nextItems, input };
|
||||
},
|
||||
{ map: new Map(), items: new Set() },
|
||||
),
|
||||
finalizeValue(({ items }) => {
|
||||
// Destroy all remaining items when no longer subscribed
|
||||
logger.debug(
|
||||
`[${name}] End of scope, destroying all ${items.size} items…`,
|
||||
);
|
||||
for (const item of items) {
|
||||
logger.debug(
|
||||
`[${name}] Destroying item with keys ${item.keys.join(", ")}`,
|
||||
);
|
||||
item.scope.end();
|
||||
}
|
||||
}),
|
||||
map(({ items, input }) =>
|
||||
project(
|
||||
[...items].map(({ item }) => item),
|
||||
input,
|
||||
),
|
||||
),
|
||||
);
|
||||
/* eslint-enable @typescript-eslint/no-explicit-any */
|
||||
}
|
||||
Reference in New Issue
Block a user