observable.ts
279 lines
| 8.8 KiB
| video/mp2t
|
TypeScriptLexer
cin
|
r138 | import { id as mid} from "module"; | ||
|
cin
|
r116 | import { Cancellation } from "@implab/core-amd/Cancellation"; | ||
|
cin
|
r138 | import { TraceSource } from "@implab/core-amd/log/TraceSource"; | ||
|
cin
|
r129 | import { isPromise } from "@implab/core-amd/safe"; | ||
|
cin
|
r153 | import { Unsubscribable, Observer, Producer, FusedSink, FusedProducer, AccumulatorFn, Observable, OperatorFn, Subscribable } from "./observable/interfaces"; | ||
|
cin
|
r96 | |||
|
cin
|
r116 | |||
|
cin
|
r102 | |||
|
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 | |||
| const noop = () => { }; | ||||
|
cin
|
r153 | |||
|
cin
|
r138 | |||
|
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()); | ||