implab/implabjs-djx Commit - r133:a3fba6b6c42e

added 'buffer' and 'subject' observable operators

cin -

r133:a3fba6b6c42e default

parent child

Context file:

r133:a3fba6b6c42e

Collapse all files

djx/src/main/ts/operators/buffer.ts created 644 +105 0

@@ -0,0 +1,105
	1	import { Cancellation } from "@implab/core-amd/Cancellation";
	2	import { CancelledError } from "@implab/core-amd/CancelledError";
	3	import { Producer, Sink, Subscribable } from "../observable";
	4	import { Scope } from "../tsx/Scope";
	5
	6	/**
	7	* Creates a buffer with the specified length. The buffer will immediately
	8	* subscribe to the source observable and start accumulating values.
	9	*
	10	* The resulting observable will repeat the buffered values for each new subscriber.
	11	*
	12	* @param length The number of elements to store.
	13	* @param ct Cancellation token to unsubscribe from the original observable.
	14	*
	15	*/
	16	export const buffer = (length: number, ct = Cancellation.none) => <T>(source: Subscribable<T>): Producer<T> => {
	17	type Status = "active" \| "complete" \| "error";
	18
	19	// ring-buffer, wpos will rotate in range (0...length-1)
	20	let wpos = 0;
	21	const buffer: T[] = [];
	22
	23	// writes the next value to the buffer
	24	const write = (value: T) => {
	25	buffer[wpos] = value;
	26	wpos = (wpos + 1) % length;
	27	};
	28
	29	// reads contents of the buffer
	30	// cb will be called for each value in the buffer
	31	const read = (cb: (item: T) => void) => {
	32	const start = wpos + length - buffer.length;
	33	const end = wpos + length;
	34
	35	for(let pos = start; pos < end; pos++ )
	36	cb(buffer[pos % length]);
	37	};
	38
	39	let status: Status = "active";
	40	let lastError: unknown;
	41	let subscribers: Sink<T>[] = [];
	42
	43	const scope = new Scope();
	44
	45	// cleanup method to release resources held by this subscription
	46	const cleanup = (cb: (item: Sink<T>) => void) => {
	47	scope.destroy();
	48	const _subscribers = subscribers;
	49	subscribers = [];
	50	_subscribers.forEach(cb);
	51	};
	52
	53	const sink: Sink<T> = {
	54	isClosed: () => status !== "active",
	55	complete: () => {
	56	if (status === "active") {
	57	status = "complete";
	58	cleanup(s => s.complete());
	59	}
	60	},
	61	error: e => {
	62	if (status === "active") {
	63	status = "error";
	64	lastError = e;
	65	cleanup(s => s.error(e));
	66	}
	67	},
	68	next: v => {
	69	if (status === "active") {
	70	write(v);
	71	const _subscribers = subscribers;
	72	_subscribers.forEach(s => s.next(v));
	73	}
	74	}
	75	};
	76
	77	if (ct.isRequested()) {
	78	sink.error(new CancelledError("The operation was cancelled", ct));
	79	} else {
	80	scope.own(source.subscribe(sink));
	81	scope.own(ct.register(e => sink.error(e)));
	82	}
	83
	84	return (s: Sink<T>) => {
	85	const _subscribers = subscribers;
	86	read(s.next);
	87	switch (status) {
	88	case "active":
	89	subscribers.push(s);
	90	return () => {
	91	if (_subscribers === subscribers) {
	92	const pos = subscribers.indexOf(s);
	93	if (pos >= 0)
	94	subscribers.splice(pos, 1);
	95	}
	96	};
	97	case "complete":
	98	s.complete();
	99	break;
	100	case "error":
	101	s.error(lastError);
	102	break;
	103	}
	104	};
	105	}; No newline at end of file

djx/src/main/ts/operators/subject.ts created 644 +50 0

@@ -0,0 +1,50
	1	import { Producer, Sink, Subscribable } from "../observable";
	2
	3	const noop = () => { };
	4
	5	/** Joins multiple subscribers to the single one. The resulting subscriber
	6	* will be created when the first client subscribes and will be released
	7	* with the the last client unsubscribed.
	8	*
	9	* Use this wrapper to prevent spawning multiple producers.
	10	*
	11	* @param source The source observable
	12	* @returns The wrapped producer
	13	*/
	14	export const subject = <T>(source: Subscribable<T>): Producer<T> => {
	15	let subscribers: Sink<T>[] = [];
	16
	17	let subscription = { unsubscribe: noop };
	18
	19	// cleanup method to release resources held by this subscription
	20	const cleanup = (cb: (item: Sink<T>) => void) => {
	21	const _subscribers = subscribers;
	22	subscribers = [];
	23	_subscribers.forEach(cb);
	24	subscription.unsubscribe();
	25	};
	26
	27	const sink: Sink<T> = {
	28	isClosed: () => false,
	29	complete: () => cleanup(s => s.complete()),
	30	error: e => cleanup(s => s.error(e)),
	31	next: v => subscribers.forEach(s => s.next(v))
	32	};
	33
	34	return client => {
	35	const _subscribers = subscribers;
	36	subscribers.push(client);
	37	if (subscribers.length === 1)
	38	subscription = source.subscribe(sink);
	39
	40	return () => {
	41	if (_subscribers === subscribers) {
	42	const pos = subscribers.indexOf(client);
	43	if (pos >= 0)
	44	subscribers.splice(pos, 1);
	45	if (!subscribers.length)
	46	subscription.unsubscribe();
	47	}
	48	};
	49	};
	50	};

.vscode/settings.json +2 -1

             {
                 "java.configuration.updateBuildConfiguration": "automatic",
                 "files.exclude": {
                     "**/.classpath": true,
                     "**/.project": true,
                     "**/.settings": true,
                     "**/.factorypath": true
                 },
                 "cSpell.words": [
                     "dijit",
                     "djbase",
                     "djclass",
-                    "Unsubscribable"
+                    "Unsubscribable",
+                    "wpos"
                 ]
             }

djx/src/main/ts/observable.ts +18 -139

             import { Cancellation } from "@implab/core-amd/Cancellation";
             import { ICancellation } from "@implab/core-amd/interfaces";
             import { isPromise } from "@implab/core-amd/safe";
             /**
              * The interface for the consumer of an observable sequence
              */
             export interface Observer<T> {
                 /**
                  * Called for the next element in the sequence
                  */
-                next: (value: T) => void;
+                next?: (value: T) => void;
                 /**
                  * Called once when the error occurs in the sequence.
                  */
-                error: (e: unknown) => void;
+                error?: (e: unknown) => void;
                 /**
                  * Called once at the end of the sequence.
                  */
-                complete: () => void;
+                complete?: () => void;
             }
             /**
              * The group of functions to feed an observable. These methods are provided to
              * the producer to generate a stream of events.
              */
             export type Sink<T> = {
                 /**
                  * Call to send the next element in the sequence
                  */
                 next: (value: T) => void;
                 /**
                  * Call to notify about the error occurred in the sequence.
                  */
                 error: (e: unknown) => void;
                 /**
                  * Call to signal the end of the sequence.
                  */
                 complete: () => void;
                 /**
                  * Checks whether the sink is accepting new elements. It's safe to
                  * send elements to the closed sink.
                  */
                 isClosed: () => boolean;
             };
             export type Producer<T> = (sink: Sink<T>) => (void | (() => void));
             export interface Unsubscribable {
                 unsubscribe(): void;
             }
             export const isUnsubscribable = (v: unknown): v is Unsubscribable =>
                 v !== null && v !== undefined && typeof (v as Unsubscribable).unsubscribe === "function";
             export const isSubscribable = <T = unknown>(v: unknown): v is Subscribable<T> =>
                 v !== null && v !== undefined && typeof (v as Subscribable<unknown>).subscribe === "function";
             export interface Subscribable<T> {
-                subscribe(consumer: Partial<Observer<T>>): Unsubscribable;
+                subscribe(consumer: Observer<T>): Unsubscribable;
             }
             export type AccumulatorFn<T, A> = (acc: A, value: T) => A;
             export type OperatorFn<T, U> = (source: Observable<T>) => Producer<U>;
             /** 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>;
                 /** 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>;
                 /** 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>;
                 /** Applies accumulator to each value in the sequence and
                  * emits the accumulated value for each source element
                  *
                  * @param accumulator
                  * @param initial
                  */
                 scan<A>(accumulator: AccumulatorFn<T, A>, initial: A): Observable<A>;
                 scan(accumulator: AccumulatorFn<T, T>): Observable<T>;
                 /** 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
                  *
                  * 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.
                  */
                 cat(...seq: Subscribable<T>[]): Observable<T>;
                 /** Pipes the specified operator to produce the new observable
                  * @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.
                  */
                 pipe<U>(op: OperatorFn<T, U>): Observable<U>;
                 /** Waits for the next event to occur and returns a promise for the next value
                  * @param ct Cancellation token
                 */
                 next(ct?: ICancellation): Promise<T>;
                 /** Collects items of the sequence to the array. */
                 collect(ct?: ICancellation): Promise<T[]>;
             }
             const noop = () => { };
-            const sink = <T>(consumer: Partial<Observer<T>>) => {
+            const sink = <T>(consumer: Observer<T>) => {
                 const { next, error, complete } = consumer;
                 return {
                     next: next ? next.bind(consumer) : noop,
                     error: error ? error.bind(consumer) : noop,
                     complete: complete ? complete.bind(consumer) : noop,
                     isClosed: () => false
                 };
             };
             /** Wraps the producer to handle tear down logic and subscription management
              *
              * @param producer The producer to wrap
              * @returns The wrapper producer
              */
             const fuse = <T>(producer: Producer<T>) => ({ next, error, complete }: Sink<T>) => {
                 let done = false;
                 let cleanup = noop;
                 const _fin = <A extends unknown[]>(fn: (...args: A) => void) =>
                     (...args: A) => done ?
                         void (0) :
                         (done = true, cleanup(), fn(...args));
                 const _fin0 = () => done ? void (0) : (done = true, cleanup());
                 const safeSink = {
                     next: (value: T) => { !done && next(value); },
                     error: _fin(error),
                     complete: _fin(complete),
                     isClosed: () => done
                 };
                 cleanup = producer(safeSink) ?? noop;
                 return done ? cleanup() : _fin0;
             };
             const _observe = <T>(producer: Producer<T>): Observable<T> => ({
-                subscribe: (consumer: Partial<Observer<T>>) => ({
+                subscribe: (consumer: Observer<T>) => ({
                     unsubscribe: producer(sink(consumer)) ?? noop
                 }),
                 map: (mapper) => _observe(({ next, ...rest }) =>
                     producer({
                         next: next !== noop ? (v: T) => next(mapper(v)) : noop,
                         ...rest
                     })
                 ),
                 filter: (predicate) => _observe(({ next, ...rest }) =>
                     producer({
                         next: next !== noop ? (v: T) => predicate(v) ? next(v) : void (0) : noop,
                         ...rest
                     })
                 ),
                 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
                     })
                 ),
                 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
                         });
                     }
                 }),
                 reduce: <A>(...args: [AccumulatorFn<T, A>, A] | [AccumulatorFn<T, T>]) => _observe<T | A>(({ next, complete, error, ...rest }) => {
                     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();
                                 }
                             },
                             error,
                             ...rest
                         });
                     } else {
                         const [accumulator, initial] = args;
                         let _acc = initial;
                         return producer({
                             next: next !== noop ? (v: T) => {
                                 _acc = accumulator(_acc, v);
                             } : noop,
                             complete: () => {
                                 next(_acc);
                                 complete();
                             },
                             error,
                             ...rest
                         });
                     }
                 }),
                 cat: (...seq) => _observe(({ next, complete: final, ...rest }) => {
                     let cleanup: () => void;
                     const complete = () => {
                         const continuation = seq.shift();
                         if (continuation) {
                             // if we have a next sequence, subscribe to it
                             const subscription = continuation.subscribe({ next, complete, ...rest });
                             cleanup = subscription.unsubscribe.bind(subscription);
                         } else {
                             // otherwise notify the consumer about completion
                             final();
                         }
                     };
                     cleanup = producer({ next, complete, ...rest }) ?? noop;
                     return () => cleanup();
                 }),
-                pipe: <U>(op: (source: Observable<T>) => Producer<U>) => observe(op(_observe(producer))),
+                pipe: <U>(op: OperatorFn<T,U>) => observe(op(_observe(producer))),
                 next: collect(
                     producer,
                     ({ next, complete, error, isClosed }) => ({
                         next: v => (next(v), complete()),
                         complete: () => error(new Error("The sequence is empty")),
                         error,
                         isClosed
                     })
                 ),
                 collect: collect(
                     producer,
                     ({ next, complete, ...rest }) => {
                         const data: T[] = [];
                         return {
                             next: v => data.push(v),
                             complete: () => (next(data), complete()),
                             ...rest
                         };
                     }
                 )
             });
             const collect = <T, U>(
                 producer: Producer<T>,
                 collector: (result: Sink<U>) => Sink<T>
             ) => (ct = Cancellation.none) => new Promise<U>((resolve, reject) => {
                 const fused = fuse<U>(({ next, complete, error, isClosed }) => {
                     const h = ct.register(error);
                     const cleanup = !isClosed() ?
                         producer(collector({ next, complete, error, isClosed })) ?? noop :
                         noop;
                     return () => {
                         h.destroy();
                         cleanup();
                     };
                 });
                 fused({
                     next: resolve,
                     error: reject,
                     complete: noop,
                     isClosed: () => false
                 });
             });
             export const observe = <T>(producer: Producer<T>) => _observe(fuse(producer));
+            /** Converts an array to the observable sequence of its elements. */
             export const ofArray = <T>(items: T[]) => _observe<T>(
                 ({ next, complete }) => (
                     items.forEach(next),
                     complete()
                 )
             );
+            /** Converts a subscribable to the observable */
+            export const ofSubscribable = <T>(subscribable: Subscribable<T>) =>
+                observe(sink => {
+                    const subscription = subscribable.subscribe(sink);
+                    return () => subscription.unsubscribe();
+                });
             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())
             );
+            /** Converts a list of parameter values to the observable sequence. The
+             * order of elements in the list will be preserved in the resulting sequence.
+             */
             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);
                     }
                 );
             export const empty = _observe<never>(({ complete }) => complete());
-            /**
-             * Creates a mutable state and the observable for the stored value.
-             * @param value The initial value for the state
-             * @returns an array of three elements `[observable, setter, getter]`
-             * The returned observable keeps the actual value and will emit it as the next
-             * element each time a consumer subscribes the observable.
-             * Calling the setter will update the stored value in the observable and notify
-             * all consumers.
-             */
-            export const stateful = <T>(producer: Producer<T>): Producer<T> => {
-                const fusedProducer = fuse(producer);
-                type Status = "active" | "complete" | "error";
-                let lastValue: T;
-                let hasValue = false;
-                let status: Status = "active";
-                let lastError: unknown;
-                let subscribers: Sink<T>[] = [];
-                const sink: Sink<T> = {
-                    isClosed: () => status !== "active",
-                    complete: () => {
-                        if (status === "active") {
-                            status = "complete";
-                            const _subscribers = subscribers;
-                            subscribers = [];
-                            _subscribers.forEach(s => s.complete());
-                    },
-                    error: e => {
-                        if (status === "active") {
-                            status = "error";
-                            lastError = e;
-                            const _subscribers = subscribers;
-                            subscribers = [];
-                            _subscribers.forEach(s => s.error(e));
-                    },
-                    next: v => {
-                        if (status === "active") {
-                            hasValue = true;
-                            lastValue = v;
-                            const _subscribers = subscribers;
-                            _subscribers.forEach(s => s.next(v));
-                };
-                fusedProducer(sink);
-                return (s: Sink<T>) => {
-                    const _subscribers = subscribers;
-                    switch (status) {
-                        case "active":
-                            if (hasValue)
-                                s.next(lastValue); // if hasValue is true,
-                            // lastValue has a valid value
-                            subscribers.push(s);
-                            return () => {
-                                if (_subscribers === subscribers) {
-                                    const pos = subscribers.indexOf(s);
-                                    if (pos >= 0)
-                                        subscribers.splice(pos, 1);
-                            };
-                        case "complete":
-                            s.complete();
-                            break;
-                        case "error":
-                            s.error(lastError);
-                            break;
-                };
-            };
-            /** Create the producer which will be called once when the first subscriber is
-             * attached, next subscribers would share the same producer. When all
-             * subscribers are removed the producer will be cleaned up.
-             * Use this wrapper to prevent spawning multiple producers.
-             * @param producer The source producer
-             * @returns The wrapped producer
-             */
-            export const subject = <T>(producer: Producer<T>): Producer<T> => {
-                const fusedProducer = fuse(producer);
-                let subscribers: Sink<T>[] = [];
-                let cleanup = noop;
-                const sink: Sink<T> = {
-                    isClosed: () => false,
-                    complete: () => {
-                        const _subscribers = subscribers;
-                        subscribers = [];
-                        _subscribers.forEach(s => s.complete());
-                        cleanup();
-                    },
-                    error: e => {
-                        const _subscribers = subscribers;
-                        subscribers = [];
-                        _subscribers.forEach(s => s.error(e));
-                        cleanup();
-                    },
-                    next: v => {
-                        const _subscribers = subscribers;
-                        _subscribers.forEach(s => s.next(v));
-                };
-                return client => {
-                    const _subscribers = subscribers;
-                    subscribers.push(client);
-                    if (subscribers.length === 1)
-                        cleanup = fusedProducer(sink) ?? noop;
-                    return () => {
-                        if (_subscribers === subscribers) {
-                            const pos = subscribers.indexOf(client);
-                            if (pos >= 0)
-                                subscribers.splice(pos, 1);
-                            if (!subscribers.length)
-                                cleanup();
-                    };
-                };
-            };
  No newline at end of file

djx/src/main/ts/tsx/render.ts +2 0

             import { TraceSource } from "@implab/core-amd/log/TraceSource";
             import { isPromise } from "@implab/core-amd/safe";
             import { id as mid } from "module";
             import { IScope, Scope } from "./Scope";
             import { isNode, isRendition, isWidget } from "./traits";
             const trace = TraceSource.get(mid);
             interface Context {
                 readonly scope: IScope;
                 readonly hooks?: (() => void)[];
             }
             let _context: Context = {
                 scope: Scope.dummy
             };
             let _renderCount = 0;
             let _renderId = 1;
             let _renderedHooks: (() => void)[] = [];
             const guard = (cb: () => unknown) => {
                 try {
                     const result = cb();
                     if (isPromise(result)) {
                         const warn = (ret: unknown) => trace.error("The callback {0} competed asynchronously. result = {1}", cb, ret);
                         result.then(warn, warn);
                     }
                 } catch (e) {
                     trace.error(e);
                 }
             };
             /**
              *
              * @param scope
              * @returns
              */
             export const beginRender = (scope = getScope()) => {
                 const prev = _context;
                 _renderCount++;
                 const renderId = _renderId++;
                 trace.debug("beginRender [{0}], pending = {1}", renderId, _renderCount);
                 if (_renderCount === 1)
                     onRendering();
                 _context = {
                     scope,
                     hooks: []
                 };
                 return endRender(prev, _context, renderId);
             };
             /**
              * Method for a deferred rendering. Returns a promise with `beginRender()` function.
              * Call to `scheduleRender` will save the current context, and will increment pending
              * operations counter.
              *
              * @example
              *
              * const begin = await scheduleRender();
              * const end = begin();
              * try {
              *     // do some DOM manipulations
              * } finally {
              *     end();
              * }
              *
              * @param scope
              * @returns
              */
             export const scheduleRender = async (scope = getScope()) => {
                 _renderCount++;
                 const renderId = _renderId ++;
                 trace.debug("scheduleRender [{0}], pending = {1}", renderId, _renderCount);
                 if (_renderCount === 1)
                     onRendering();
                 await Promise.resolve();
                 return () => {
                     trace.debug("beginRender [{0}], pending = {1}", renderId, _renderCount);
                     const prev = _context;
                     _context = {
                         scope,
                         hooks: []
                     };
                     return endRender(prev, _context, renderId);
                 };
             };
             /**
              * Completes render operation
              */
             const endRender = (prev: Context, current: Context, renderId: number) => () => {
                 if (_context !== current)
                     trace.error("endRender mismatched beginRender call");
                 const { hooks } = _context;
                 if (hooks)
                     hooks.forEach(guard);
                 _renderCount--;
                 _context = prev;
                 trace.debug("endRender [{0}], pending = {1}", renderId, _renderCount);
                 if (_renderCount === 0)
                     onRendered();
             };
             // called when the first beginRender is called for this iteration
             const onRendering = () => {
+                trace.log("Rendering started");
                 setTimeout(() => {
                     if (_renderCount !== 0)
                         trace.error("Rendering tasks aren't finished, currently running = {0}", _renderCount);
                 });
             };
             // called when all render operations are complete
             const onRendered = () => {
+                trace.log("Rendering compete");
                 _renderedHooks.forEach(guard);
                 _renderedHooks = [];
             };
             export const whenRendered = () => new Promise<void>((resolve) => {
                 if (_renderCount)
                     _renderedHooks.push(resolve);
                 else
                     resolve();
             });
             export const renderHook = (hook: () => void) => {
                 const { hooks } = _context;
                 if (hooks)
                     hooks.push(hook);
                 else
                     guard(hook);
             };
             export const refHook = <T>(value: T, ref: JSX.Ref<T>) => {
                 const { hooks, scope } = _context;
                 if (hooks)
                     hooks.push(() => ref(value));
                 else
                     guard(() => ref(value));
                 scope.own(() => ref(undefined));
             };
             /** Returns the current scope */
             export const getScope = () => _context.scope;
             /** Schedules the rendition to be rendered to the DOM Node
              * @param rendition The rendition to be rendered
              * @param scope The scope
              */
             export const render = (rendition: unknown, scope = Scope.dummy) => {
                 const complete = beginRender(scope);
                 try {
                     return getItemDom(rendition);
                 } finally {
                     complete();
                 }
             };
             const emptyFragment = document.createDocumentFragment();
             /** Renders DOM element for different types of the argument. */
             export const getItemDom = (v: unknown) => {
                 if (typeof v === "string" || typeof v === "number" || v instanceof RegExp || v instanceof Date) {
                     // primitive types converted to the text nodes
                     return document.createTextNode(v.toString());
                 } else if (isNode(v)) {
                     // nodes are kept as is
                     return v;
                 } else if (isRendition(v)) {
                     // renditions are instantiated
                     return v.getDomNode();
                 } else if (isWidget(v)) {
                     // widgets are converted to it's markup
                     return v.domNode;
                 } else if (typeof v === "boolean" || v === null || v === undefined) {
                     // null | undefined | boolean are removed
                     return emptyFragment;
                 } else if (v instanceof Array) {
                     // arrays will be translated to document fragments
                     const fragment = document.createDocumentFragment();
                     v.map(item => getItemDom(item))
                         .forEach(node => fragment.appendChild(node));
                     return fragment;
                 } else {
                     // bug: explicit error otherwise
                     throw new Error(`Invalid parameter: ${String(v)}`);
                 }
             };

djx/src/test/ts/state-tests.ts +4 -3

-            import { observe, stateful } from "./observable";
+            import { observe } from "./observable";
+            import { buffer } from "./operators/buffer";
             import * as t from "tap";
             interface CounterState {
                 count: number;
                 label: "low" | "mid" | "high"
             }
             let set: (v: CounterState) => void = () => void (0);
             const initial: CounterState = { count: 0, label: "low" };
             let value = initial;
-            const obs = observe(stateful<CounterState>(({ next }) => {
+            const obs = observe<CounterState>(({ next }) => {
                 next(initial);
                 set = next;
-            }));
+            }).pipe(buffer(2));
             set({ count: 10, label: "mid" });
             obs.subscribe({
                 next: v => value = v
             });
             t.equal(value.count, 10, "State should update");
             set({ count: 20, label: "high" });

djx/src/test/ts/subject-tests.ts +6 -5

-            import { observe, subject } from "./observable";
+            import { observe } from "./observable";
+            import { subject } from "./operators/subject";
             import * as tap from "tap";
             tap.test("Subject tests", t => {
                 let nextEvent: (value: string) => void = () => void (0);
-                const subj1 = observe(subject<string>(({ next }) => {
+                const subj1 = observe<string>(({ next }) => {
                     t.comment("Start subject");
                     nextEvent = next;
                     return () => {
                         nextEvent = () => void (0);
                         t.comment("Stop subject");
                     };
-                }));
+                }).pipe(subject);
                 const h1 = subj1.subscribe({
                     next: v => t.comment(`h1 next: ${v}`)
                 });
                 nextEvent("first");
                 const h2 = subj1.subscribe({
                     next: v => t.comment(`h2 next: ${v}`)
                 });
                 nextEvent("second");
                 h1.unsubscribe();
                 nextEvent("third");
                 h2.unsubscribe();
                 t.pass("Subject finished");
                 t.end();
             }).catch(e => console.error(e));
             tap.test("Subject tests #2", t => {
                 let nextEvent: (value: string) => void = () => void (0);
-                const subj1 = observe(subject<string>(({ next, complete }) => {
+                const subj1 = observe<string>(({ next, complete }) => {
                     t.comment("Start subject");
                     complete();
                     nextEvent = next;
                     return () => {
                         nextEvent = () => void (0);
                         t.comment("Stop subject");
                     };
-                }));
+                }).pipe(subject);
                 const h1 = subj1.subscribe({
                     next: v => t.comment(`h1 next: ${v}`)
                 });
                 nextEvent("first");
                 const h2 = subj1.subscribe({
                     next: v => t.comment(`h2 next: ${v}`)
                 });
                 nextEvent("second");
                 h1.unsubscribe();
                 nextEvent("third");
                 h2.unsubscribe();
                 t.pass("Subject finished");
                 t.end();
             }).catch(e => console.error(e));

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