implab/implabjs-djx Commit - r157:f9518367061a

WIP observables

cin -

r157:f9518367061a default

parent child

Context file:

r157:f9518367061a

Collapse all files

djx/src/main/ts/observable.ts +4 -40

@@ -1,5 +1,5
1	import { isPromise } from "@implab/core-amd/safe";	1	import { isPromise } from "@implab/core-amd/safe";
2	import { Unsubscribable, Producer, ~~FusedSink~~, Observable, Subscribable } from "./observable/interfaces";	2	import { Unsubscribable, Producer, Observable, Subscribable } from "./observable/interfaces";
3	import { ObservableImpl } from "./observable/ObservableImpl";	3	import { ObservableImpl } from "./observable/ObservableImpl";
4		4
5	export const isUnsubscribable = (v: unknown): v is Unsubscribable =>	5	export const isUnsubscribable = (v: unknown): v is Unsubscribable =>
@@ -8,46 +8,10 export const isUnsubscribable = (v: unkn
8	export const isSubscribable = <T = unknown>(v: unknown): v is Subscribable<T> =>	8	export const isSubscribable = <T = unknown>(v: unknown): v is Subscribable<T> =>
9	v !== null && v !== undefined && typeof (v as Subscribable<unknown>).subscribe === "function";	9	v !== null && v !== undefined && typeof (v as Subscribable<unknown>).subscribe === "function";
10		10
11	const noop = () => { };	11	export const observe = <T>(producer: Producer<T>): Observable<T> => new ObservableImpl<T>(producer);
12
13	/** Wraps the producer to handle tear down logic and subscription management
14	*
15	* The resulting producer will invoke cleanup logic on error or complete events
16	* and will prevent calling of any method from the sink.
17	*
18	* @param producer The producer to wrap
19	* @returns The wrapper producer
20	*/
21	const fuse = <T>(producer: Producer<T>) => ({ next, error, complete }: FusedSink<T>) => {
22	let done = false;
23	let cleanup = noop;
24
25	const _fin = <A extends unknown[]>(fn: (...args: A) => void) =>
26	(...args: A) => done ?
27	void (0) :
28	(done = true, cleanup(), fn(...args));
29
30	const _fin0 = () => done ? void (0) : (done = true, cleanup());
31
32	const safeSink = {
33	next: (value: T) => { !done && next(value); },
34	error: _fin(error),
35	complete: _fin(complete),
36	isClosed: () => done
37	};
38	// call the producer
39	cleanup = producer(safeSink) ?? noop;
40	// if the producer throws exception bypass it to the caller rather then to
41	// the sink. This is a feature.
42
43	// if the producer completed the sequence immediately call the cleanup in place
44	return done ? cleanup() : _fin0;
45	};
46
47	export const observe = <T>(producer: Producer<T>): Observable<T> => new ObservableImpl<T>(fuse(producer));
48		12
49	/** Converts an array to the observable sequence of its elements. */	13	/** Converts an array to the observable sequence of its elements. */
50	export const ofArray = <T>(items: T[]) => new ObservableImpl<T>(	14	export const ofArray = <T>(items: T[]): Observable<T> => new ObservableImpl<T>(
51	({ next, complete }) => (	15	({ next, complete }) => (
52	items.forEach(next),	16	items.forEach(next),
53	complete()	17	complete()
@@ -100,6 +64,6 export const of = <T>(...items: (T \| Pro
100	}	64	}
101	);	65	);
102		66
103	export const empty = new ObservableImpl<never>(({ complete }) => complete());	67	export const empty: Observable<never> = new ObservableImpl<never>(({ complete }) => complete());
104		68
105	export type * from "./observable/interfaces"; No newline at end of file	69	export type * from "./observable/interfaces";

djx/src/main/ts/observable/ObservableImpl.ts +69 -31

             import { id as mid } from "module";
-            import { AccumulatorFn, Observable, Observer, OperatorFn, Producer, Subscribable } from "./interfaces";
+            import { AccumulatorFn, Observable, Observer, OperatorFn, Producer, Sink, Subscribable, Unsubscribable } from "./interfaces";
             import { TraceSource } from "@implab/core-amd/log/TraceSource";
             import { Predicate } from "@implab/core-amd/interfaces";
             import { Cancellation } from "@implab/core-amd/Cancellation";
             const errorFallback = (e: unknown) => trace.error("Unhandled observable error: {0}", e);
-            const subscription = () => {
+            const subscription = (reject: (err: unknown) => void, ct = Cancellation.none) => {
                 let done = false;
                 let cleanup = noop;
                             cb();
                         } else {
                             const _prev = cleanup;
-                            cleanup = () => (_prev(), cb());
+                            cleanup = _prev === noop ? cb : () => (_prev(), cb());
                         }
                     }
                 };
-                return { finalize, isClosed, teardown };
+                const error = (err: unknown) => finalize() && reject(err);
+                if (ct.isSupported()) {
+                    const h = ct.register(error);
+                    teardown(() => h.destroy());
+                }
+                return { finalize, isClosed, teardown, error };
             };
+            type FusedSink<T> = Omit<Sink<T>, "isClosed">;
+            export const fuse = <T>(producer: Producer<T>) => (consumer: FusedSink<T>) => {
+                const reject = (err: unknown) => consumer.error ? consumer.error(err) : errorFallback(err);
+                const { teardown, isClosed, finalize, error } = subscription(reject);
+                const next = consumer.next ? consumer.next.bind(consumer) : noop;
+                teardown(producer({
+                    next: next !== noop ? (v => isClosed() || next(v)) : noop,
+                    error,
+                    complete: () => finalize() && consumer.complete && consumer.complete(),
+                    isClosed
+                }));
+                return { unsubscribe: finalize };
+            };
+            type FusedProducer<T> = (consumer: FusedSink<T>) => void | (() => void);
             export class ObservableImpl<T> implements Observable<T> {
-                private readonly _producer: Producer<T>;
+                private readonly _producer: FusedProducer<T>;
-                constructor(producer: Producer<T>) {
+                constructor(producer: FusedProducer<T>) {
                     this._producer = producer;
                 }
                 subscribe(consumer: Observer<T> = {}) {
-                    const { teardown, isClosed, finalize } = subscription();
+                    const reject = (err: unknown) => consumer.error ? consumer.error(err) : errorFallback(err);
+                    const next = consumer.next ? consumer.next.bind(consumer) : noop;
+                    const { teardown, isClosed, finalize, error } = subscription(reject);
-                    teardown(this._producer({
+                    teardown((0, this._producer)({
-                        next: consumer.next ? consumer.next.bind(consumer) : noop,
+                        next: next !== noop ? (v => isClosed() || next(v)) : noop,
-                        error: err => finalize() && (consumer.error ? consumer.error(err) : errorFallback(err)),
+                        error,
-                        complete: () => finalize() && consumer.complete && consumer.complete(),
+                        complete: () => finalize() && consumer.complete && consumer.complete()
-                        isClosed
                     }));
                     return { unsubscribe: finalize };
                 }
                 tap({ next: tapNext, complete: tapComplete, error: tapError }: Observer<T>) {
-                    return new ObservableImpl<T>(({ next, complete, error, isClosed }) =>
+                    return new ObservableImpl<T>(({ next, complete, error }) =>
                         this._producer({
                             next: tapNext ? (v => (tapNext(v), next(v))) : next,
                             complete: tapComplete ? (() => (tapComplete(), complete())) : complete,
-                            error: tapError ? (e => (tapError(e), error(e))) : error,
+                            error: tapError ? (e => (tapError(e), error(e))) : error
-                            isClosed
                         }));
                 }
                     );
                 }
+                scan<A>(accumulator: AccumulatorFn<T, A>, initial: A): Observable<A>;
+                scan(accumulator: AccumulatorFn<T, T>): Observable<T>;
                 scan<A>(...args: [AccumulatorFn<T, A>, A] | [AccumulatorFn<T, T>]) {
                     if (args.length === 1) {
                         return new ObservableImpl<T | A>(({ next, ...rest }) => {
                     }
                 }
+                reduce<A>(accumulator: AccumulatorFn<T, A>, initial: A): Observable<A>;
+                reduce(accumulator: AccumulatorFn<T, T>): Observable<T>;
                 reduce<A>(...args: [AccumulatorFn<T, A>, A] | [AccumulatorFn<T, T>]) {
                     if (args.length === 1) {
-                        return new ObservableImpl<T | A>(({ next, complete, error, isClosed }) => {
+                        return new ObservableImpl<T | A>(({ next, complete, error }) => {
                             const [accumulator] = args;
                             let _acc: T;
                             let index = 0;
                                         complete();
                                     }
                                 },
-                                error,
+                                error
-                                isClosed
                             });
                         });
                     } else {
-                        return new ObservableImpl<T | A>(({ next, complete, error, isClosed }) => {
+                        return new ObservableImpl<T | A>(({ next, complete, error }) => {
                             const [accumulator, initial] = args;
                             let _acc = initial;
                             return this._producer({
                                     next(_acc);
                                     complete();
                                 },
-                                error,
+                                error
-                                isClosed
                             });
                         });
                     }
                     return new Promise<T>((resolve, reject) => {
                         ct.throwIfRequested();
-                        const { teardown, finalize, isClosed } = subscription();
+                        const { teardown, finalize, isClosed, error } = subscription(reject, ct);
-                        const error = (err: unknown) => finalize() && reject(err);
-                        const h = ct.register(error);
-                        teardown(() => h.destroy());
                         teardown((0, this._producer)({
                             next: (item: T) => finalize() && resolve(item),
                             complete: () => finalize() && reject(new Error("The sequence is empty")),
                         const acc: T[] = [];
-                        const { teardown, finalize, isClosed } = subscription();
+                        const { teardown, finalize, isClosed, error } = subscription(reject, ct);
-                        const error = (err: unknown) => finalize() && reject(err);
-                        const h = ct.register(error);
-                        teardown(() => h.destroy());
                         teardown((0, this._producer)({
                             next: (item: T) => isClosed() ? void (0) : acc.push(item),
                             complete: () => finalize() && resolve(acc),
                     });
                 }
+                forEach(next: (item: T) => void, ct = Cancellation.none): Promise<void> {
+                    return new Promise<void>((resolve, reject) => {
+                        ct.throwIfRequested();
+                        const { teardown, finalize, isClosed, error } = subscription(reject, ct);
+                        teardown((0, this._producer)({
+                            next: item => !isClosed() && next(item),
+                            complete: () => finalize() && resolve(),
+                            error,
+                            isClosed
+                        }));
+                    });
+                }
             }

djx/src/main/typings/observable/interfaces.d.ts +12 -4

@@ -52,10 +52,6 export type Sink<T> = {
52		52
53	export type Producer<T> = (sink: Sink<T>) => (void \| (() => void));	53	export type Producer<T> = (sink: Sink<T>) => (void \| (() => void));
54		54
55	export type FusedSink<T> = Omit<Sink<T>, "isClosed">;
56
57	export type FusedProducer<T> = (sink: FusedSink<T>) => (void \| (() => void));
58
59	export interface Unsubscribable {	55	export interface Unsubscribable {
60	unsubscribe(): void;	56	unsubscribe(): void;
61	}	57	}
@@ -153,4 +149,16 export interface Observable<T> extends S
153		149
154	/** Collects items of the sequence to the array. */	150	/** Collects items of the sequence to the array. */
155	collect(ct?: ICancellation): Promise<T[]>;	151	collect(ct?: ICancellation): Promise<T[]>;
		152
		153	/**
		154	* Iterates through the elements in this observable until the end of the
		155	* sequence is reached or the operation is cancelled.
		156	*
		157	* @param next The callback for the next item in the sequence
		158	* @param ct The optional cancellation token for this operation
		159	*
		160	* @returns A promise which will be fulfilled when the operation is completed.
		161	* In case of the cancellation this promise will be rejected.
		162	*/
		163	forEach(next: (item: T) => void, ct?: ICancellation): Promise<void>;
156	} No newline at end of file	164	}

djx/src/test/ts/observable-tests.ts +4 -1

             import { Cancellation } from "@implab/core-amd/Cancellation";
             import { delay } from "@implab/core-amd/safe";
-            const subj1 = observe<number>(({ next, complete }) => {
+            const subj1 = observe<number>(({ next, error, complete }) => {
                 next(1);
                 complete();
                 next(2);
+                next(3);
+                error(new Error("This error should be ignored"));
+                next(4);
             });
             const consumer1 = {

General Comments 0

Write
Preview

You need to be logged in to leave comments. Login now

No TODOs yet

	Site-wide shortcuts
/	Use quick search box
g h	Goto home page
g g	Goto my private gists page
g G	Goto my public gists page
g 0-9	Goto bookmarked items from 0-9
n r	New repository page
n g	New gist page

	Repositories
g s	Goto summary page
g c	Goto changelog page
g f	Goto files page
g F	Goto files page with file search activated
g p	Goto pull requests page
g o	Goto repository settings
g O	Goto repository access permissions settings
t s	Toggle sidebar on some pages