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

              ## SYNOPSIS
-             ```tsx
+             ```jsx
              import { djbase, djclass, bind, prototype, AbstractConstructor } from "@implab/djx/declare";
              import { DjxWidgetBase } from "@implab/djx/tsx/DjxWidgetBase";
              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() {
                      "experimentalDecorators": true,
                      "jsxFactory": "createElement",
                      "jsx": "react",
-                     "target": "ES5", // minimal supported version
+                     "target": "ES5",
                      "lib": ["ES2015", "DOM"]
                  }
              }
              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";
              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"
              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>
              * 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
                  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 = () => { };
                  pipe: <U>(op: (source: Observable<T>) => Producer<U>) => observe(op(_observe(producer))),
-                 next: (ct?: ICancellation) => {
-                     const _ct = ct ?? Cancellation.none;
-                     return new Promise<T>((resolve, reject) => {
-                         // 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: collect(
+                     producer,
+                     ({ next, complete, error, isClosed }) => ({
-                                     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 () => {
-                             };
-                         });
-                         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>(
-                 ({ next, error, complete }) => void promise.then(
+             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())
              );
-             export const of = <T>(...items: T[]) => _observe<T>(
-                 ({ next, complete }) => (
-                     items.forEach(next),
-                     complete()
+                 )
+             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());

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

              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 })))
-                                     .then(undefined, error);
+                                 results.then(processResults).then(undefined, error);
                              } else {
-                                 results.forEach((item, newIndex) => next({ item, newIndex, prevIndex: -1 }));
+                                 processResults(results);
                              }
                              if (!isClosed() && (options?.observe !== false) && isDjObservableResults<T>(results)) {

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 {
                      type Ref<T> = ((value: T | undefined) => void);
-                     type Element = Rendition;
+                     type Element = Rendition | _WidgetBase;
                      interface DjxIntrinsicAttributes<E> {
                          /** alias for className */

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

-             import { empty, observe } from "./observable";
-             import * as t from "tap";
+             import { empty, observe, of } from "./observable";
+             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);
              };
              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,
              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");
-             t.equal(maps, 3, "The map chain should not be executed after completion");
-             t.ok(consumer2.completed, "The completion signal should pass through");
+             tap.equal(consumer2.value, 8, "Should map");
+             tap.equal(maps, 3, "The map chain should not be executed after completion");
+             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");
                  };
              });
              subj3.subscribe(consumer3);
-             t.equal(consumer3.odd, 2, "Should get 2 odd elements");
-             t.equal(consumer3.even, 2, "Should get 2 even elements");
-             t.ok(consumer3.completed, "The sequence should completed");
-             t.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.odd, 2, "Should get 2 odd elements");
+             tap.equal(consumer3.even, 2, "Should get 2 even elements");
+             tap.ok(consumer3.completed, "The sequence should completed");
+             tap.equal(consumer3.subscribed, 1, "The subscription should be done once");
+             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),
-                 complete: () => t.comment("subj2: complete")
+                 next: val => tap.comment("subj2: reduce =", val),
+                 complete: () => tap.comment("subj2: complete")
              });
-             subj2.reduce((a,b) => a + b).next()
-                 .then(value => t.comment("subj2: next reduce=", value))
-                 .catch(() => {});
+             tap.test("of(...) tests", async t => {
+                 await subj2.reduce((a, b) => a + b).next()
+                     .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