implab/implabjs-djx Commit - r129:66546e709732

added observable.collect() method to collect a sequnce to the array...

cin -

r129:66546e709732 v1.8.0 default

parent child

Context file:

r129:66546e709732

Collapse all files

djx/readme.md +37 -6

             # @implab/djx
             ## SYNOPSIS
-            ```tsx
+            ```jsx
             import { djbase, djclass, bind, prototype, AbstractConstructor } from "@implab/djx/declare";
             import { DjxWidgetBase } from "@implab/djx/tsx/DjxWidgetBase";
             import { createElement } from "@implab/djx/tsx";
             interface MyWidgetAttrs {
                 title: string;
                 counter: number;
             }
             interface MyWidgetEvents {
                 "count-inc": Event;
                 "count-dec": Event;
             }
             @djclass
             export class MyWidget extends djbase(
                 DjxWidgetBase as AbstractConstructor<DjxWidgetBase<MyWidgetAttrs, MyWidgetEvents>>
             ) {
                 @bind({ node: "titleNode", type: "innerHTML" })
                 title = "";
                 @prototype()
                 counter = 0;
                 render() {
                     const Frame = (props: any) => <div>{props.children}</div>;
                     return <div
                         className="myWidget"
                         tabIndex={3}
                         style={ alignContent: "center", border: "1px solid" }
                     >
                         <h1 data-dojo-attach-point="titleNode"></h1>
                         <Frame>
                             <span class="up-button" onclick={e => this._onIncClick(e)}>[+]</span>
                             <span class="down-button" onclick={() => this._onDecClick()}>[-]</span>
                         </Frame>
                     </div>;
                 }
                 _onIncClick(e: MouseEvent) {
                     this.emit("count-inc", { bubbles: false });
                 }
                 _onDecClick() {
                     this.emit("count-dec", { bubbles: false });
                 }
             }
             ```
             ## DESCRIPTION
             This package provides you with the tools to glue your good-fellow dojo with modern
             techniques of building the webapp. The core concept is to built around widgets and
             using .tsx to write it. Here are some features:
             * `djbase()`, `@djaclass` - traits to declare your classes with `dojo/_base/declare`
             * `@implab/djx/tsx` - traits to build the rendering of your widgets with tsx
             * `DjxWidgetBase` - abstract class which supports tsx markup and
               `data-dojo-attach-*` attributes.
             * `@bind(...)` - annotations provide an easy way of using standard dojo widget
               attribute bindings.
             ### djbase, @djclass
             These two traits provide convenient way of using `dojo/_base/declare` in Typescript
             for declaring your classes.
             `djbase(...constructors)` - this method accepts a list of constructors in its
             parameters and returns the **fake** base type which then can be used to derive
             your own class. This allows you to provide the Typescript with the correct
             information about the base type and even use `super`!. The only caveat of
             this approach is that you **MUST** decorate your class with `@djclass` annotation.
             Consider the following example:
             ```ts
             import { djbase, djclass } from "@implab/djx/declare";
             import { FooMixin } from "./FooMixin";
             import { BarMixin } from "./BarMixin";
             import { BoxMixin } from "./BoxMixin";
             @djclass
             export class Baz extends djbase(FooMixin, BarMixin, BoxMixin) {
                 writeHello(out: string[]) {
                     out.push("-> Baz");
                     super.writeHello(out);
                     out.push("<- Baz");
                 }
             }
             ```
             All mixins are declared like the one below:
             ```ts
             import { djclass, djbase } from "@implab/djx/declare";
             interface Super {
                 writeHello(out: string[]): void;
             }
             @djclass
             export class BarMixin extends djbase<Super>() {
                 writeHello(out: string[]) {
                     out.push("-> Bar");
                     super.writeHello(out);
                     out.push("<- Bar");
                 }
             }
             ```
             finally create an instance and call the `writeHello` method
             ```ts
             const baz = new Baz();
             const data: string[] = [];
             baz.writeHello(data);
             console.log(data.join("\n"));
             ```
             you will get the following output:
             ```text
             -> Baz
             -> Box
             -> Bar
             -> Foo
             <- Foo
             <- Bar
             <- Box
             <- Baz
             ```
             Let's take a closer look at the `Baz` declaration it uses `djbase` to derive
             from three mixins and the class is decorated with `@djclass` to accomplish the
             declaration and make a real constructor.
             To allow access to the next sibling method (in terms of multiple inheritance)
             Dojo provides `this.inherited(arguments)` method but this approach leads to the
             problem with 'strict' mode of ES5 and eliminates the type information about a
             calling method. This library solves the problem calling inherited/next method by
             utilizing `super` keyword. Under the hood there are proxy methods generated in
             the prototype of the declared class which make calls to `this.inherited(...)`
             method. This technique is compatible with 'strict' mode.
             Mixins are declared similar, they also may have the base types although
             the most common case is declaring the mixin without any base classes. To allow
             the mixin to access the next method declare the interface with
             desired methods and use the special form of `djbase<Super>()` without arguments.
             ### DjxWidgetBase<Attrs, Events>
             This is the base class for the djx widgets. It declares the abstract method
             `render()` which is used to render the content of the widget, like `_TemplatedMixin`.
             This class extends `dijit/_WidgetBase` and contains logic from `_AttachMixin` thus
             it is capable to handle `data-dojo-attach-*` attributes from the rendered markup.
-            ```tsx
+            ```jsx
             @djclass
             export class MyFirstWidget extends djbase(DjxWidgetBase) {
                 render() {
                     return <h1>My first widget</h1>;
                 }
             }
             ```
             ### Markup (.tsx)
             Add to your `tsconfig.json` the following options
             ```json
             {
                 "compilerOptions": {
                     "types": [
                         "@implab/djx",
                         "@implab/dojo-typings"
                     ],
                     "skipLibCheck": true,
                     "experimentalDecorators": true,
                     "jsxFactory": "createElement",
                     "jsx": "react",
-                    "target": "ES5", // minimal supported version
+                    "target": "ES5",
                     "lib": ["ES2015", "DOM"]
                 }
             }
             ```
             Import `createElement` into your `.tsx` file
             ```ts
             import { createElement } from "@implab/djx/tsx";
             ```
             You are ready to go!
             ### Adding reactive behavior: refs, watch(...) and watchFor(...)
             This library adds some reactive traits to update the generated DOM of the widget.
             Dojo 1.x adds some standard options to deal with dynamic changes:
             * `data-dojo-attach-point` allows to get reference to an element (or a nested widget)
             * widget attribute mappings, allows to bind widget's property to a property of
               the element, referenced by `data-dojo-attach-point`.
             The typical implementation of this technique would look like
-            ```tsx
+            ```jsx
             import { createElement } from "@implab/djx/tsx";
             import {djclass, djbase, bind} from "@implab/djx/declare";
             @djclass
             export class MyFirstWidget extends djbase(DjxWidgetBase) {
                 // @bind will generate special attribute mapping
                 // _setCaptionAttr = { node: "captionNode", type: "innerHTML" }
                 @bind({ node: "captionNode", type: "innerHTML" })
                 caption = "My first widget";
                 render() {
                     return <h1 data-dojo-attach-point="captionNode"/>;
                 }
             }
             ```
             Despite this is a natural way for the dojo it has some disadvantages:
 . The compiler doesn't check existence of the attach-point.
 . Attribute mappings support only simple mappings, it's difficult to update the
                complex rendition.
             This library helps you to get both goals with special trait `watch(...)`
-            ```tsx
+            ```jsx
             import { createElement } from "@implab/djx/tsx";
             import { djclass, djbase} from "@implab/djx/declare"
             @djclass
             export class MyFirstWidget extends djbase(DjxWidgetBase) {
                 caption = "My first widget";
                 render() {
                     return <h1>{watch(this,"caption", value => value)}</h1>;
                 }
             }
             ```
             In this example we replaced attach-point with simple call to `watch` function
             which renders string value to text representation (text node). It will create a
             rendition which will observe the `caption` property of the widget and update its
             contents according to the value changes of the property.
             The key feature of this approach that the rendering function within `watch` may
             return a complex rendition.
-            ```tsx
+            ```jsx
             // inside some widget
             render() {
                 return <section>
                     {watch(this,"user", value => value && [
                         <UserInfo user={value}/>,
                         <LogoutButton click={this._logoutClick}/>
                     ])}
                 </section>;
             }
             private readonly _logoutClick = () => { /* do logout */ }
             ```
             The `watch` function has two forms:
             * `watch(stateful, prop, render)` - observes the specified property of the
               `dojo/Stateful` object (or widget)
             * `watch(observable, render)` - observes the specified observable. It supports
                `rxjs` or `@implab/djx/observable` observables.
             The `render` callback may return almost anything which will be converted to DOM:
             * `boolean`, `null`, `undefined` - ignored,
             * `string` - converted to text node,
             * `array` - converted to DocumentFragment of its elements,
             * DOM Nodes and widgets are left intact,
             * any other kind of value will cause an error.
+            The watch method allows to observe a single value, for the large sets of data
+            this isn't suitable well and may lead to performance issues. Dojo provides
+            observable stores to being able to track individual changes. The library provides
+            `watchFor(observable, render)` method to render observable query results and
+            handle changes on per item basis.
+            ```jsx
+            // inside some widget
+            staff = new Observable(new Memory<Employee>()),
+            getStuff() {
+                return this.staff.query();
+            }
+            addEmployee(employee: Employee) {
+                this.staff.add(employee); // the rendition will update automatically
+            }
+            render() {
+                return <table>
+                    <thead>
+                        <tr><th>Name</th><th>Position</th><th>Salary</th></tr>
+                    </thead>
+                    <tbody>
+                    {watchFor(this.getStaff(), ({name, position, salary}) =>
+                        <tr><td>{name}</td><td>{position}</td><td>{salary}</td></tr>
+                    )}
+                    </tbody>
+                </table>
+            }
+            ```

djx/src/main/ts/observable.ts +64 -24

             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;
                 /**
                  * Called once when the error occurs in the sequence.
                  */
                 error: (e: unknown) => void;
                 /**
                  * Called once at the end of the sequence.
                  */
                 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;
             }
             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 to
+                 * @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 { 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>>) => ({
                     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))),
-                next: (ct?: ICancellation) => {
+                next: collect(
-                    const _ct = ct ?? Cancellation.none;
+                    producer,
-                    return new Promise<T>((resolve, reject) => {
+                    ({ next, complete, error, isClosed }) => ({
-                        // wrap the producer to handle only single event
-                        const once = fuse<T>(({ next, complete, error, isClosed }) => {
-                            const h = _ct.register(error);
-                            // is the _ct fires it will call error() and isClosed() will return true
-                            const cleanup = !isClosed() ?
-                                producer({
                         next: v => (next(v), complete()),
                         complete: () => error(new Error("The sequence is empty")),
                         error,
                         isClosed
-                                }) ?? noop :
+                    })
+                ),
+                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();
                     };
                 });
-                        once({
+                fused({
                     next: resolve,
                     error: reject,
                     complete: noop,
                     isClosed: () => false
                 });
             });
-            });
             export const observe = <T>(producer: Producer<T>) => _observe(fuse(producer));
-            export const streamArray = <T>(items: T[]) => _observe<T>(
+            export const ofArray = <T>(items: T[]) => _observe<T>(
                 ({ next, complete }) => (
                     items.forEach(next),
                     complete()
                 )
             );
-            export const streamPromise = <T>(promise: PromiseLike<T>) => observe<T>(
+            const of1 = <T>(item: T | PromiseLike<T>) => observe<T>(
-                ({ next, error, complete }) => void promise.then(
+                ({ next, error, complete }) =>
+                    isPromise(item) ?
+                        void item.then(
                             v => (next(v), complete()),
                             error
+                        ) :
+                        (next(item), complete())
             );
-            export const of = <T>(...items: T[]) => _observe<T>(
+            export const of = <T>(...items: (T | PromiseLike<T>)[]) => items.length === 1 ?
-                ({ next, complete }) => (
+                of1(items[0]) :
-                    items.forEach(next),
+                observe<T>(
-                    complete()
+                    ({ 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();
                         }
                     };
                 };
             };

djx/src/main/ts/store.ts +6 -3

             import { PromiseOrValue } from "@implab/core-amd/interfaces";
             import { isPromise } from "@implab/core-amd/safe";
             import { observe, Observable } from "./observable";
             export interface OrderedUpdate<T> {
                 /** The item is being updated */
                 readonly item: T;
                 /** The previous index of the item, -1 in case it is inserted */
                 readonly prevIndex: number;
                 /** The new index of the item, -1 in case it is deleted */
                 readonly newIndex: number;
             }
             export type QueryResults<T> = Observable<OrderedUpdate<T>>;
             interface DjObservableResults<T> {
                 /**
                  * Allows observation of results
                  */
                 observe(listener: (object: T, previousIndex: number, newIndex: number) => void, includeUpdates?: boolean): {
                     remove(): void;
                 };
             }
             interface Queryable<T, Q, O> {
                 query(query?: Q, options?: O): PromiseOrValue<T[]>;
             }
             export const isDjObservableResults = <T>(v: object): v is DjObservableResults<T> =>
                 v && (typeof (v as { observe?: unknown; }).observe === "function");
             export const query = <T, Q, O>(store: Queryable<T, Q, O>, includeUpdates = true) =>
                 (query?: Q, options?: O & { observe?: boolean }) => {
                     return observe<OrderedUpdate<T>>(({ next, complete, error, isClosed }) => {
+                        const processResults = (items: T[]) =>
+                            items.forEach((item, newIndex) => next({ item, newIndex, prevIndex: -1 }));
                         try {
                             const results = store.query(query, options);
                             if (isPromise(results)) {
-                                results.then(items => items.forEach((item, newIndex) => next({ item, newIndex, prevIndex: -1 })))
+                                results.then(processResults).then(undefined, error);
-                                    .then(undefined, error);
                             } else {
-                                results.forEach((item, newIndex) => next({ item, newIndex, prevIndex: -1 }));
+                                processResults(results);
                             }
                             if (!isClosed() && (options?.observe !== false) && isDjObservableResults<T>(results)) {
                                 const h = results.observe((item, prevIndex, newIndex) => next({ item, prevIndex, newIndex }), includeUpdates);
                                 return () => h.remove();
                             } else {
                                 complete();
                             }
                         } catch (err) {
                             error(err);
                         }
                     });
                 };

djx/src/main/typings/index.d.ts +2 -1

             // eslint-disable-next-line @typescript-eslint/triple-slash-reference
             /// <reference path="./css-plugin.d.ts"/>
+            import _WidgetBase = require("dijit/_WidgetBase");
             import { Rendition } from "./tsx/traits";
             declare global {
                 namespace JSX {
                     type Ref<T> = ((value: T | undefined) => void);
-                    type Element = Rendition;
+                    type Element = Rendition | _WidgetBase;
                     interface DjxIntrinsicAttributes<E> {
                         /** alias for className */
                         class?: string;
                         /** specifies the name of the property in the widget where the the
                          * reference to the current object will be stored
                          */
                         "data-dojo-attach-point"?: string;
                         /** specifies handlers map for the events */
                         "data-dojo-attach-event"?: string;
                         ref?: Ref<E>;
                         /** @deprecated */
                         [attr: string]: unknown;
                     }
                     // eslint-disable-next-line @typescript-eslint/no-empty-interface
                     interface DjxIntrinsicElements {
                     }
                     type RecursivePartial<T> = T extends string | number | boolean | null | undefined | ((...args: unknown[]) => unknown) ?
                         T :
                         { [k in keyof T]?: RecursivePartial<T[k]> };
                     type MatchingMemberKeys<T, U> = {
                         [K in keyof T]: T[K] extends U ? K : never;
                     }[keyof T];
                     type NotMatchingMemberKeys<T, U> = {
                         [K in keyof T]: T[K] extends U ? never : K;
                     }[keyof T];
                     type ExtractMembers<T, U> = Pick<T, MatchingMemberKeys<T, U>>;
                     type ExcludeMembers<T, U> = Pick<T, NotMatchingMemberKeys<T, U>>;
                     type ElementAttrNames<E> = NotMatchingMemberKeys<E, (...args: unknown[]) => unknown>;
                     type ElementAttrType<E, K extends string | symbol> = K extends keyof E ? RecursivePartial<E[K]> : string;
                     type ElementAttrNamesBlacklist = "children" | "getRootNode" | keyof EventTarget;
                     /** This type extracts keys of the specified parameter E by the following rule:
                      *  1. skips all ElementAttrNamesBlacklist
                      *  2. skips all methods except with the signature of event handlers
                      */
                     type AssignableElementAttrNames<E> = {
                         [K in keyof E]: K extends ElementAttrNamesBlacklist ? never :
                         ((evt: Event) => unknown) extends E[K] ? K :
                         E[K] extends ((...args: unknown[]) => unknown) ? never :
                         K;
                     }[keyof E];
                     type LaxElement<E extends object> =
                         RecursivePartial<Pick<E, AssignableElementAttrNames<E>>> &
                         DjxIntrinsicAttributes<E>;
                     type LaxIntrinsicElementsMap = {
                         [tag in keyof HTMLElementTagNameMap]: LaxElement<HTMLElementTagNameMap[tag]>
                     } & DjxIntrinsicElements;
                     type IntrinsicElements = {
                         [tag in keyof LaxIntrinsicElementsMap]: LaxIntrinsicElementsMap[tag];
                     };
                     interface ElementChildrenAttribute {
                         children: unknown;
                     }
                     interface IntrinsicClassAttributes<T> {
                         ref?: Ref<T>;
                         children?: unknown;
                     }
                 }
             }

djx/src/test/ts/observable-tests.ts +47 -24

-            import { empty, observe } from "./observable";
+            import { empty, observe, of } from "./observable";
-            import * as t from "tap";
+            import * as tap from "tap";
             import { Cancellation } from "@implab/core-amd/Cancellation";
+            import { delay } from "@implab/core-amd/safe";
             const subj1 = observe<number>(({ next, complete }) => {
                 next(1);
                 complete();
                 next(2);
             });
             const consumer1 = {
                 sum: 0,
                 next(v: number) {
                     this.sum += v;
                 }
             };
             subj1.subscribe(consumer1);
-            t.equal(consumer1.sum, 1, "Should get only one value");
+            tap.equal(consumer1.sum, 1, "Should get only one value");
             subj1.subscribe(consumer1);
-            t.equal(consumer1.sum, 2, "Should get the value again");
+            tap.equal(consumer1.sum, 2, "Should get the value again");
             const consumer2 = {
                 value: 0,
                 completed: false,
                 next(v: number) { this.value = v; },
                 complete() { this.completed = true; }
             };
             let maps = 0;
             subj1
                 .map(v => {
-                    t.comment(`map1: ${v * 2}`);
+                    tap.comment(`map1: ${v * 2}`);
                     maps++;
                     return v * 2;
                 })
                 .map(v => {
-                    t.comment(`map2: ${v * 2}`);
+                    tap.comment(`map2: ${v * 2}`);
                     maps++;
                     return v * 2;
                 })
                 .map(v => {
-                    t.comment(`map3: ${v * 2}`);
+                    tap.comment(`map3: ${v * 2}`);
                     maps++;
                     return v * 2;
                 })
                 .subscribe(consumer2);
-            t.equal(consumer2.value, 8, "Should map");
+            tap.equal(consumer2.value, 8, "Should map");
-            t.equal(maps, 3, "The map chain should not be executed after completion");
+            tap.equal(maps, 3, "The map chain should not be executed after completion");
-            t.ok(consumer2.completed, "The completion signal should pass through");
+            tap.ok(consumer2.completed, "The completion signal should pass through");
             const subj2 = observe<number>(({ next, complete }) => {
                 [1, 2, 3, 4, 5].forEach(next);
                 complete();
                 return () => {
-                    t.comment("subj2: unsubscribe");
+                    tap.comment("subj2: unsubscribe");
                 };
             });
             const consumer3 = {
                 even: 0,
                 odd: 0,
                 completed: false,
                 subscribed: 0,
                 unsubscribed: 0,
                 next(v: "even" | "odd") {
                     this[v]++;
                 },
                 complete() {
                     this.completed = true;
                 },
                 subscribe() {
                     this.subscribed++;
                 },
                 unsubscribe() {
                     this.unsubscribed++;
                 }
             };
             const subj3 = subj2.pipe<"even" | "odd">(self => ({ next, complete, error }) => {
                 consumer3.subscribe();
                 let count = 0;
                 const h = self.subscribe({
                     next: val => {
                         if (val % 2 === 0)
                             next("odd");
                         else
                             next("even");
                         if (++count === 4)
                             complete();
                     },
                     complete,
                     error
                 });
                 return () => {
                     consumer3.unsubscribe();
                     h.unsubscribe();
                 };
             });
             subj3.subscribe(consumer3);
-            t.equal(consumer3.odd, 2, "Should get 2 odd elements");
+            tap.equal(consumer3.odd, 2, "Should get 2 odd elements");
-            t.equal(consumer3.even, 2, "Should get 2 even elements");
+            tap.equal(consumer3.even, 2, "Should get 2 even elements");
-            t.ok(consumer3.completed, "The sequence should completed");
+            tap.ok(consumer3.completed, "The sequence should completed");
-            t.equal(consumer3.subscribed, 1, "The subscription should be done once");
+            tap.equal(consumer3.subscribed, 1, "The subscription should be done once");
-            t.equal(consumer3.unsubscribed, 1, "The cleanup should be done after completion");
+            tap.equal(consumer3.unsubscribed, 1, "The cleanup should be done after completion");
             subj2.reduce((a, b) => a + b).subscribe({
-                next: val => t.comment("subj2: reduce =", val),
+                next: val => tap.comment("subj2: reduce =", val),
-                complete: () => t.comment("subj2: complete")
+                complete: () => tap.comment("subj2: complete")
             });
-            subj2.reduce((a,b) => a + b).next()
+            tap.test("of(...) tests", async t => {
-                .then(value => t.comment("subj2: next reduce=", value))
+                await subj2.reduce((a, b) => a + b).next()
-                .catch(() => {});
+                    .then(value => t.comment("subj2: next reduce=", value));
-            subj2.next().then(val => t.equal(val, 1, "Should peek the first element")).catch(() => {});
+                await subj2.next().then(val => t.equal(val, 1, "Should peek the first element"));
                 const cancelled = new Cancellation(cancel => cancel());
-            t.rejects(subj2.next(cancelled), "Cancelled next() method should fail").catch(() => {});
+                await t.rejects(subj2.next(cancelled), "Cancelled next() method should fail");
+                await t.rejects(empty.next(), "Empty sequence should fail to get next element");
+                await of(delay(1).then(() => 1), Promise.resolve(2), 3)
+                    .reduce<number[]>((a, x) => [...a, x], [])
+                    .next()
+                    .then(res => t.same(res, [1, 2, 3], "of(...) should keep the order"));
+                const rejected = Promise.reject("DIE!");
+                rejected.catch(() => { }); // SAFE AND SOUND
-            t.rejects(empty.next(), "Empty sequence should fail to get next element").catch(() => {});
  No newline at end of file
+                await t.resolves(
+                    of(Promise.resolve(1), rejected).next(),
+                    "of(...) should emit non-rejected items"
+                );
+                await t.rejects(
+                    of(1, Promise.reject("DIE!")).reduce((a) => a).next(),
+                    "of(...) should terminate with error when a parameter is rejected"
+                );
+                t.same(await of(1,2,3).collect(), [1,2,3], ".collect() should return the collected sequence");
+                await t.rejects(of(1,2,3).collect(cancelled), ".collect() should support cancellation");
+            }).catch(() => { });
  No newline at end of file

djx/src/test/ts/plan.ts +1 -1

             import "./declare-tests";
             import "./observable-tests";
             import "./state-tests";
-            import "./subject-tests";
  No newline at end of file
+            import "./subject-tests";

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