observable.ts
429 lines
| 13.6 KiB
| video/mp2t
|
TypeScriptLexer
cin
|
r138 | import { id as mid} from "module"; | ||
|
cin
|
r116 | import { Cancellation } from "@implab/core-amd/Cancellation"; | ||
| import { ICancellation } from "@implab/core-amd/interfaces"; | ||||
|
cin
|
r138 | import { TraceSource } from "@implab/core-amd/log/TraceSource"; | ||
|
cin
|
r129 | import { isPromise } from "@implab/core-amd/safe"; | ||
|
cin
|
r110 | |||
|
cin
|
r138 | const trace = TraceSource.get(mid); | ||
|
cin
|
r102 | /** | ||
| * The interface for the consumer of an observable sequence | ||||
| */ | ||||
| export interface Observer<T> { | ||||
| /** | ||||
| * Called for the next element in the sequence | ||||
| */ | ||||
|
cin
|
r142 | next?(value: T): void; | ||
|
cin
|
r96 | |||
|
cin
|
r102 | /** | ||
| * Called once when the error occurs in the sequence. | ||||
| */ | ||||
|
cin
|
r142 | error?(e: unknown): void; | ||
|
cin
|
r102 | |||
| /** | ||||
| * Called once at the end of the sequence. | ||||
| */ | ||||
|
cin
|
r142 | complete?(): void; | ||
|
cin
|
r96 | } | ||
|
cin
|
r102 | /** | ||
|
cin
|
r110 | * The group of functions to feed an observable. These methods are provided to | ||
|
cin
|
r102 | * the producer to generate a stream of events. | ||
| */ | ||||
| export type Sink<T> = { | ||||
|
cin
|
r110 | /** | ||
| * Call to send the next element in the sequence | ||||
| */ | ||||
|
cin
|
r116 | next: (value: T) => void; | ||
| /** | ||||
| * Call to notify about the error occurred in the sequence. | ||||
| */ | ||||
| error: (e: unknown) => void; | ||||
|
cin
|
r110 | |||
|
cin
|
r116 | /** | ||
| * Call to signal the end of the sequence. | ||||
| */ | ||||
| complete: () => void; | ||||
|
cin
|
r110 | |||
|
cin
|
r116 | /** | ||
| * Checks whether the sink is accepting new elements. It's safe to | ||||
| * send elements to the closed sink. | ||||
| */ | ||||
| isClosed: () => boolean; | ||||
|
cin
|
r102 | }; | ||
|
cin
|
r96 | |||
| export type Producer<T> = (sink: Sink<T>) => (void | (() => void)); | ||||
|
cin
|
r142 | type FusedSink<T> = Omit<Sink<T>, "isClosed">; | ||
| type FusedProducer<T> = (sink: FusedSink<T>) => (void | (() => void)); | ||||
|
cin
|
r102 | export interface Unsubscribable { | ||
| unsubscribe(): void; | ||||
| } | ||||
|
cin
|
r118 | export const isUnsubscribable = (v: unknown): v is Unsubscribable => | ||
|
cin
|
r102 | v !== null && v !== undefined && typeof (v as Unsubscribable).unsubscribe === "function"; | ||
|
cin
|
r118 | export const isSubscribable = <T = unknown>(v: unknown): v is Subscribable<T> => | ||
|
cin
|
r102 | v !== null && v !== undefined && typeof (v as Subscribable<unknown>).subscribe === "function"; | ||
|
cin
|
r96 | |||
|
cin
|
r102 | export interface Subscribable<T> { | ||
|
cin
|
r144 | /** Subscribes a consumer to events. If a consumer isn't specified | ||
| * this method activates the producer to achieve side affects if any. | ||||
| */ | ||||
| subscribe(consumer?: Observer<T>): Unsubscribable; | ||||
|
cin
|
r102 | } | ||
|
cin
|
r116 | export type AccumulatorFn<T, A> = (acc: A, value: T) => A; | ||
|
cin
|
r136 | export type OperatorFn<T, U> = (source: Observable<T>) => Observable<U>; | ||
|
cin
|
r124 | |||
|
cin
|
r102 | /** The observable source of items. */ | ||
| export interface Observable<T> extends Subscribable<T> { | ||||
| /** Transforms elements of the sequence with the specified mapper | ||||
| * | ||||
| * @param mapper The mapper used to transform the values | ||||
| */ | ||||
| map<T2>(mapper: (value: T) => T2): Observable<T2>; | ||||
|
cin
|
r96 | |||
|
cin
|
r144 | /** Injects the specified observer into the each producer to consumer chain. | ||
| * The method is used to add side effect to the events processing. | ||||
| * | ||||
| * @param observer The consumer for the events | ||||
| */ | ||||
| tap(observer: Observer<T>): Observable<T>; | ||||
|
cin
|
r102 | /** Filters elements of the sequence. The resulting sequence will | ||
| * contain only elements which match the specified predicate. | ||||
| * | ||||
| * @param predicate The filter predicate. | ||||
| */ | ||||
| filter(predicate: (value: T) => boolean): Observable<T>; | ||||
|
cin
|
r96 | |||
|
cin
|
r124 | /** Completes the sequence once the condition is met. | ||
| * @param predicate The condition which should be met to complete the sequence | ||||
| */ | ||||
| until(predicate: (value: T) => boolean): Observable<T>; | ||||
| /** Keeps the sequence running while elements satisfy the condition. | ||||
| * | ||||
| * @param predicate The condition which should be met to continue. | ||||
| */ | ||||
| while(predicate: (value: T) => boolean): Observable<T>; | ||||
|
cin
|
r102 | /** Applies accumulator to each value in the sequence and | ||
| * emits the accumulated value for each source element | ||||
| * | ||||
| * @param accumulator | ||||
| * @param initial | ||||
| */ | ||||
|
cin
|
r116 | scan<A>(accumulator: AccumulatorFn<T, A>, initial: A): Observable<A>; | ||
| scan(accumulator: AccumulatorFn<T, T>): Observable<T>; | ||||
|
cin
|
r110 | |||
|
cin
|
r116 | /** Applies accumulator to each value in the sequence and | ||
| * emits the accumulated value at the end of the sequence | ||||
| * | ||||
| * @param accumulator | ||||
| * @param initial | ||||
| */ | ||||
| reduce<A>(accumulator: AccumulatorFn<T, A>, initial: A): Observable<A>; | ||||
| reduce(accumulator: AccumulatorFn<T, T>): Observable<T>; | ||||
| /** Concatenates the specified sequences with this observable | ||||
| * | ||||
| * @param seq sequences to concatenate with the current observable | ||||
|
cin
|
r124 | * | ||
| * The concatenation doesn't accumulate values from the specified sequences, | ||||
| * The result of the concatenation is the new observable which will switch | ||||
| * to the next observable after the previous one completes. Values emitted | ||||
| * before the next observable being active are lost. | ||||
|
cin
|
r116 | */ | ||
|
cin
|
r110 | cat(...seq: Subscribable<T>[]): Observable<T>; | ||
|
cin
|
r114 | |||
|
cin
|
r124 | |||
|
cin
|
r116 | /** Pipes the specified operator to produce the new observable | ||
|
cin
|
r124 | * @param op The operator consumes this observable and produces a new one | ||
| * | ||||
| * The operator is a higher order function which takes a source observable | ||||
| * and returns a producer for the new observable. | ||||
| * | ||||
| * This function can be used to create a complex mapping between source and | ||||
| * resulting observables. The operator may have a state (or a side effect) | ||||
| * and can be connected to multiple observables. | ||||
|
cin
|
r116 | */ | ||
|
cin
|
r124 | pipe<U>(op: OperatorFn<T, U>): Observable<U>; | ||
|
cin
|
r116 | |||
| /** Waits for the next event to occur and returns a promise for the next value | ||||
|
cin
|
r129 | * @param ct Cancellation token | ||
|
cin
|
r116 | */ | ||
| next(ct?: ICancellation): Promise<T>; | ||||
|
cin
|
r129 | |||
| /** Collects items of the sequence to the array. */ | ||||
| collect(ct?: ICancellation): Promise<T[]>; | ||||
|
cin
|
r102 | } | ||
| const noop = () => { }; | ||||
|
cin
|
r138 | const errorFallback = (e: unknown) => trace.error("Unhandled observable error: {0}", e); | ||
|
cin
|
r133 | const sink = <T>(consumer: Observer<T>) => { | ||
|
cin
|
r142 | // eslint-disable-next-line @typescript-eslint/unbound-method | ||
|
cin
|
r102 | const { next, error, complete } = consumer; | ||
|
cin
|
r96 | return { | ||
|
cin
|
r102 | next: next ? next.bind(consumer) : noop, | ||
|
cin
|
r142 | error: error ? error.bind(consumer) : errorFallback, // report unhandled errors | ||
| complete: complete ? complete.bind(consumer) : noop | ||||
|
cin
|
r109 | }; | ||
|
cin
|
r102 | }; | ||
|
cin
|
r110 | /** Wraps the producer to handle tear down logic and subscription management | ||
| * | ||||
|
cin
|
r142 | * The resulting producer will invoke cleanup logic on error or complete events | ||
| * and will prevent calling of any method from the sink. | ||||
| * | ||||
|
cin
|
r110 | * @param producer The producer to wrap | ||
| * @returns The wrapper producer | ||||
| */ | ||||
|
cin
|
r142 | const fuse = <T>(producer: Producer<T>) => ({ next, error, complete }: FusedSink<T>) => { | ||
|
cin
|
r102 | let done = false; | ||
|
cin
|
r110 | let cleanup = noop; | ||
| const _fin = <A extends unknown[]>(fn: (...args: A) => void) => | ||||
| (...args: A) => done ? | ||||
| void (0) : | ||||
| (done = true, cleanup(), fn(...args)); | ||||
|
cin
|
r124 | const _fin0 = () => done ? void (0) : (done = true, cleanup()); | ||
|
cin
|
r110 | const safeSink = { | ||
|
cin
|
r109 | next: (value: T) => { !done && next(value); }, | ||
|
cin
|
r110 | error: _fin(error), | ||
| complete: _fin(complete), | ||||
| isClosed: () => done | ||||
|
cin
|
r109 | }; | ||
|
cin
|
r142 | // call the producer | ||
|
cin
|
r110 | cleanup = producer(safeSink) ?? noop; | ||
|
cin
|
r142 | // if the producer throws exception bypass it to the caller rather then to | ||
| // the sink. This is a feature. | ||||
| // if the producer completed the sequence immediately call the cleanup in place | ||||
|
cin
|
r124 | return done ? cleanup() : _fin0; | ||
|
cin
|
r109 | }; | ||
|
cin
|
r96 | |||
|
cin
|
r142 | const _observe = <T>(producer: FusedProducer<T>): Observable<T> => ({ | ||
|
cin
|
r144 | subscribe: (consumer: Observer<T> = {}) => ({ | ||
|
cin
|
r102 | unsubscribe: producer(sink(consumer)) ?? noop | ||
| }), | ||||
|
cin
|
r116 | |||
|
cin
|
r110 | map: (mapper) => _observe(({ next, ...rest }) => | ||
|
cin
|
r102 | producer({ | ||
| next: next !== noop ? (v: T) => next(mapper(v)) : noop, | ||||
|
cin
|
r110 | ...rest | ||
| }) | ||||
| ), | ||||
|
cin
|
r116 | |||
|
cin
|
r144 | tap: ({next: tapNext, complete: tapComplete, error: tapError}) => _observe(({next,complete, error}) => | ||
| producer({ | ||||
| next: tapNext ? (v => (tapNext(v), next(v))) : next, | ||||
| complete: tapComplete ? (() => (tapComplete(), complete())): complete, | ||||
| error: tapError ? (e => (tapError(e), error(e))) : error | ||||
| }) | ||||
| ), | ||||
|
cin
|
r110 | filter: (predicate) => _observe(({ next, ...rest }) => | ||
| producer({ | ||||
| next: next !== noop ? (v: T) => predicate(v) ? next(v) : void (0) : noop, | ||||
| ...rest | ||||
|
cin
|
r102 | }) | ||
| ), | ||||
|
cin
|
r116 | |||
|
cin
|
r124 | until: predicate => _observe(({ next, complete, ...rest }) => | ||
| producer({ | ||||
| next: v => predicate(v) ? complete() : next(v), | ||||
| complete, | ||||
| ...rest | ||||
| }) | ||||
| ), | ||||
| while: predicate => _observe(({ next, complete, ...rest }) => | ||||
| producer({ | ||||
| next: v => predicate(v) ? next(v) : complete(), | ||||
| complete, | ||||
| ...rest | ||||
| }) | ||||
| ), | ||||
|
cin
|
r116 | scan: <A>(...args: [AccumulatorFn<T, A>, A] | [AccumulatorFn<T, T>]) => _observe<T | A>(({ next, ...rest }) => { | ||
| if (args.length === 1) { | ||||
| const [accumulator] = args; | ||||
| let _acc: T; | ||||
| let index = 0; | ||||
| return producer({ | ||||
| next: next !== noop ? (v: T) => next(index++ === 0 ? _acc = v : _acc = accumulator(_acc, v)) : noop, | ||||
| ...rest | ||||
| }); | ||||
| } else { | ||||
| const [accumulator, initial] = args; | ||||
| let _acc = initial; | ||||
| return producer({ | ||||
| next: next !== noop ? (v: T) => next(_acc = accumulator(_acc, v)) : noop, | ||||
| ...rest | ||||
| }); | ||||
| } | ||||
| }), | ||||
|
cin
|
r142 | reduce: <A>(...args: [AccumulatorFn<T, A>, A] | [AccumulatorFn<T, T>]) => _observe<T | A>(({ next, complete, error }) => { | ||
|
cin
|
r116 | if (args.length === 1) { | ||
| const [accumulator] = args; | ||||
| let _acc: T; | ||||
| let index = 0; | ||||
| return producer({ | ||||
| next: next !== noop ? (v: T) => { | ||||
| _acc = index++ === 0 ? v : accumulator(_acc, v); | ||||
| } : noop, | ||||
| complete: () => { | ||||
| if (index === 0) { | ||||
| error(new Error("The sequence can't be empty")); | ||||
| } else { | ||||
| next(_acc); | ||||
| complete(); | ||||
| } | ||||
| }, | ||||
|
cin
|
r142 | error | ||
|
cin
|
r116 | }); | ||
| } else { | ||||
| const [accumulator, initial] = args; | ||||
| let _acc = initial; | ||||
| return producer({ | ||||
| next: next !== noop ? (v: T) => { | ||||
| _acc = accumulator(_acc, v); | ||||
| } : noop, | ||||
| complete: () => { | ||||
| next(_acc); | ||||
| complete(); | ||||
| }, | ||||
|
cin
|
r142 | error | ||
|
cin
|
r116 | }); | ||
| } | ||||
|
cin
|
r110 | }), | ||
| cat: (...seq) => _observe(({ next, complete: final, ...rest }) => { | ||||
| let cleanup: () => void; | ||||
|
cin
|
r136 | const len = seq.length; | ||
| const complete = (i: number) => i < len ? | ||||
| () => { | ||||
| const subscription = seq[i].subscribe({ next, complete: complete(i + 1), ...rest }); | ||||
|
cin
|
r110 | cleanup = subscription.unsubscribe.bind(subscription); | ||
|
cin
|
r136 | } : final; | ||
|
cin
|
r110 | |||
|
cin
|
r136 | cleanup = producer({ next, complete: complete(0), ...rest }) ?? noop; | ||
|
cin
|
r110 | |||
| return () => cleanup(); | ||||
|
cin
|
r114 | }), | ||
|
cin
|
r136 | pipe: <U>(op: OperatorFn<T, U>) => op(_observe(producer)), | ||
|
cin
|
r110 | |||
|
cin
|
r129 | next: collect( | ||
| producer, | ||||
|
cin
|
r142 | ({ next, complete, error }) => ({ | ||
|
cin
|
r129 | next: v => (next(v), complete()), | ||
| complete: () => error(new Error("The sequence is empty")), | ||||
|
cin
|
r142 | error | ||
|
cin
|
r129 | }) | ||
| ), | ||||
| collect: collect( | ||||
| producer, | ||||
|
cin
|
r142 | ({ next, complete, error}) => { | ||
|
cin
|
r129 | const data: T[] = []; | ||
| return { | ||||
| next: v => data.push(v), | ||||
| complete: () => (next(data), complete()), | ||||
|
cin
|
r142 | error | ||
|
cin
|
r129 | }; | ||
| } | ||||
| ) | ||||
| }); | ||||
|
cin
|
r110 | |||
|
cin
|
r129 | const collect = <T, U>( | ||
|
cin
|
r142 | producer: FusedProducer<T>, | ||
| collector: (result: FusedSink<U>) => FusedSink<T> | ||||
|
cin
|
r129 | ) => (ct = Cancellation.none) => new Promise<U>((resolve, reject) => { | ||
| const fused = fuse<U>(({ next, complete, error, isClosed }) => { | ||||
| const h = ct.register(error); | ||||
| const cleanup = !isClosed() ? | ||||
|
cin
|
r142 | producer(collector({ next, complete, error })) ?? noop : | ||
|
cin
|
r129 | noop; | ||
|
cin
|
r110 | |||
|
cin
|
r129 | return () => { | ||
| h.destroy(); | ||||
| cleanup(); | ||||
| }; | ||||
| }); | ||||
|
cin
|
r110 | |||
|
cin
|
r129 | fused({ | ||
| next: resolve, | ||||
| error: reject, | ||||
|
cin
|
r142 | complete: noop | ||
|
cin
|
r129 | }); | ||
|
cin
|
r116 | }); | ||
|
cin
|
r110 | |||
| export const observe = <T>(producer: Producer<T>) => _observe(fuse(producer)); | ||||
|
cin
|
r133 | /** Converts an array to the observable sequence of its elements. */ | ||
|
cin
|
r129 | export const ofArray = <T>(items: T[]) => _observe<T>( | ||
|
cin
|
r116 | ({ next, complete }) => ( | ||
| items.forEach(next), | ||||
| complete() | ||||
| ) | ||||
| ); | ||||
|
cin
|
r110 | |||
|
cin
|
r133 | /** Converts a subscribable to the observable */ | ||
| export const ofSubscribable = <T>(subscribable: Subscribable<T>) => | ||||
|
cin
|
r142 | observe<T>(sink => { | ||
|
cin
|
r133 | const subscription = subscribable.subscribe(sink); | ||
| return () => subscription.unsubscribe(); | ||||
| }); | ||||
|
cin
|
r129 | const of1 = <T>(item: T | PromiseLike<T>) => observe<T>( | ||
| ({ next, error, complete }) => | ||||
| isPromise(item) ? | ||||
| void item.then( | ||||
| v => (next(v), complete()), | ||||
| error | ||||
| ) : | ||||
| (next(item), complete()) | ||||
|
cin
|
r116 | ); | ||
|
cin
|
r110 | |||
|
cin
|
r133 | /** Converts a list of parameter values to the observable sequence. The | ||
| * order of elements in the list will be preserved in the resulting sequence. | ||||
| */ | ||||
|
cin
|
r129 | export const of = <T>(...items: (T | PromiseLike<T>)[]) => items.length === 1 ? | ||
| of1(items[0]) : | ||||
| observe<T>( | ||||
| ({ next, error, complete, isClosed }) => { | ||||
| const n = items.length; | ||||
| const _next = (start: number) => { | ||||
| if (start > 0 && isClosed()) // when resumed | ||||
| return; | ||||
| for (let i = start; i < n; i++) { | ||||
| const r = items[i]; | ||||
| if (isPromise(r)) { | ||||
| r.then(v => (next(v), _next(i + 1)), error); | ||||
| return; // suspend | ||||
| } else { | ||||
| next(r); | ||||
| } | ||||
| } | ||||
| complete(); | ||||
| }; | ||||
| _next(0); | ||||
| } | ||||
| ); | ||||
|
cin
|
r110 | |||
|
cin
|
r124 | export const empty = _observe<never>(({ complete }) => complete()); | ||