implab/implabjs-djx Commit - r154:2a5720a0816e

Merge

cin -

r154:2a5720a0816e default

parent child

Context file:

r154:2a5720a0816e

Collapse all files

.hgtags +1 0

             fc9f82c082ef432137da086a1fe9c37a12ba16a2 v1.0.0-rc2
 d80d7901b4c8ffe8728e4a7bf5f4b7e7a669bb5 v1.0.0-rc3
 a2c44d8e1f34dd30c2b50f92b7dc2e8f3247c43 v1.0.0-rc5
 c01fabe9ea9fb5e753fbeae8b0d2664e7072a66 v1.0.0-rc6
 e546fe36fdddc8324f1098ee950fa1a7ba19b93 v1.0.0-rc7
 f4d5e2c719a20ae6d65f1f4b5e2141ed765e975 v1.0.0-rc8
             a1ab2b5975ad4b19599fb61538e7aaf329fb528c v1.0.0-rc10
 b77ac3bf8f200876450ad50c308a9441a7f39c7 v1.0.0-rc11
 b72f33756d3d10553b743f0b9f4504148cf97d v1.0.0-rc12
             b88fac0e76c0e61e397e2995f468f7cf342afbc9 v1.0.0-rc13
             a46488b209e8aac583c1634043147d87740c63b4 v1.0.0-rc14
             1174538197f6796384e643f62100292f1377b137 v1.0.0-rc15
             e8012fdf09ae442094f3831abe70649f8520659e 1.0.0-rc16
             a1a1ef050ecc9d2a780e308cbffc2ad915a5c13d 1.0.0-rc17
 c6c7e16919cff4019a55661725789d287439b75 v1.0.0-rc18
 b6c4159c66cecf6c5957d33ad474919608489c6 v1.0.0
 b2dcc1ae97fe3673242c84fe8d4b50a55f0 v1.0.1
             ed3c20c09b000386b5204b483955eb61ee662eff v1.0.2
 ea350f98bb7a06e636251ea5cad403217e868 v1.0.3
 ba910a5425ed95f7a67ef4db1dbbf599e2cab v1.0.5
             deb0ed6fb68015d89ba1287f6b766d1f80b6e4ff v1.0.7
             dd0d589acfbbe938a5630547e83ad934ab597b64 v1.0.8
             f2499237b5bf0ac2d73a8e437b8f15fa8a6306a5 v1.0.9
             d4f0cdae9577caa605d9317d0ceb93f7b67895f3 v1.0.10
 a0018655d1c3dafc16ae50971cb535b8555e246 v1.1.0
             ff3695b0a48f00155ad6c3d7e000fe185ee6f95a v1.1.1
             c276b9b7fa833242d1bf75eeeaae9924feee51e8 v1.1.2
             fdde09e66c009f4950ec69a393a8e789725da758 v1.2.0
 c6055f20ce1a49bdca5e6dabce88ee8deed v1.2.1
             bc7556143fe536e3df374bc0070146311884284e v1.2.2
             e5bb5e80ce96fc4ca0fbbf0b5390443a616e99fa v1.2.3
             cc5be30e84f8a0d9a8e25447651576d9e46ab154 v1.2.4
 ab11174c7446830d91d5f1b652a18c6ecae5 v1.2.5
 a7b6319ebe24973fe10b8d82c19d3d86857b4e v1.2.6
 deb750dc18fcd9be83c28cb8371530fffd8 v1.2.7
 f8caa5bf8fd2d4a56a6cff40be31ab82a2727 v1.2.8
 c829aa08a2cf65cc074640199e79c5e49d2c93 v1.3.0
 a190b3a757dc650c2c612073cbffaaa720b6090 v1.4.0
 ccfaae984e9a458580664923c87f258cff9890f v1.4.4
             e9a9ed6d7647848e9c7f28e6e9f92d009259607d v1.5.0
             aac297dda27dac1e86bc93617ba172a2c92415f5 v1.6.0
             e07418577cbcaa6e31ccbf87c4495324ecbd88bd v1.6.1
             bc1b4dd8ca1a03af095ea1a61169c6025ef37fce v1.6.2
             fb2ea4d6aabac2b374f3b2bfddaca6a83dedf34f v1.6.3
             cede47727a1b0867d19be1d168aceb00b5e74e3f v1.7.0
 aad89415099d8e1942a8e67050c22e111d8b v1.7.1
 e70973248de01601aa364eadf11dd2f3a9f v1.8.0
             c7d9ad82b374c2a20710eba8fa40a9b1472ddc77 v1.8.1
             f139e2153e0da5649558b1ed9369da860d442c0d v1.9.0-rc1
 ce00ba2452d565f1ed3bb884643b4b88b9cde v1.9.0-rc2
 b2c550c676ff6acf16f5bd56ee3804342f80b7 v1.9.0-rc3
 d1b83ebdfaa93599451175055a94f711c079f v1.9.0-rc4
 b8239b953c0384cf32cab46cf41dac97ea03b v1.9.0-rc5
 d91ae4b711eeb2145b9685240e1afada904 v1.9.0-rc6
             af4f8424e83d56e89a64f39e19514ca10dbd43c6 v1.9.0
 f3ad8e6cffcf32fc5a555e55e8544227293d4c v1.10.0
+            d9c99ae7dec87ef47aded05def6a094a892df0e6 v1.10.1

djx/src/main/ts/store.ts +24 -5

             import { PromiseOrValue } from "@implab/core-amd/interfaces";
             import { isCancellable, isPromise } from "@implab/core-amd/safe";
             import { observe, Observable, empty } from "./observable";
             import { after } from "dojo/aspect";
             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) => {
                 const q = queryEx(store, includeUpdates);
                 return (query?: Q, options?: O & { observe?: boolean }) => {
                     const [data, updates] = q(query, options);
                     return options?.observe === false ? data : data.cat(updates);
                 };
             };
+            /**
+             * Wraps the query method of the store, the resulting method takes a query
+             * expression and returns two observable sequences. The first sequence represents
+             * the results of the query, the second sequence provides the updates to the
+             * query results.
+             *
+             * @param store The store used to query data
+             * @param includeUpdates The flag to include item updates not only additions and
+             *  deletions. By default this flag is set to true.
+             * @returns Two observable sequences
+             */
             export const queryEx = <T, Q, O>(store: Queryable<T, Q, O>, includeUpdates = true) =>
                 (query?: Q, options?: O): [data: QueryResults<T>, updates: QueryResults<T>] => {
-                    const pending: T[] = [];
+                    /** count active observers */
+                    let listeners = 0;
-                    let results: PromiseOrValue<T[]> = pending;
+                    let results: PromiseOrValue<T[]> = [];
                     const data = observe<OrderedUpdate<T>>(({ next, complete, error }) => {
                         const processResults = (items: T[]) =>
                             items.forEach((item, newIndex) => next({ item, newIndex, prevIndex: -1 }));
                         try {
-                            if (results === pending)
+                            // is there are no active observers here, we need to query actual
+                            // data from the store.
+                            if (listeners === 0)
                                 results = store.query(query, options);
                             if (isPromise(results)) {
                                 results.then(processResults).then(complete, error);
                                 if (isCancellable(results))
                                     return results.cancel.bind(results);
                             } else {
                                 processResults(results);
                                 complete();
                             }
                         } catch (e) {
                             error(e);
                         }
                     });
                     const updates = observe<OrderedUpdate<T>>(({ next, complete, error, isClosed }) => {
                         try {
                             if (!isClosed() && isDjObservableResults<T>(results)) {
+                                // subscribe fot the changes
+                                listeners++;
                                 const h = results.observe((item, prevIndex, newIndex) => next({ item, prevIndex, newIndex }), includeUpdates);
-                                return () => h.remove();
+                                return () => {
+                                    // unsubscribe from changes
+                                    listeners--;
+                                    h.remove();
+                                };
                             } else {
                                 complete();
                             }
                         } catch (e) {
                             error(e);
                         }
                     });
                     return [data, updates];
                 };
             interface IndexedStore<T> {
                 get(id: string | number): PromiseLike<T> | T | null | undefined;
             }
             interface Notifications<T> {
                 notify(item: T | undefined, id: string | number | undefined): void;
             }
             const hasNotifications = <T>(x: unknown): x is Notifications<T> =>
                 typeof x === "object" && x !== null && (typeof (x as Notifications<T>).notify === "function");
             interface GetOpts {
                 observe?: boolean;
             }
             export type ItemUpdate<T> = [item: NonNullable<T>, id: string | number | undefined] |
             [item: undefined | null, id: string | number];
             const filterItem = (itemId: string | number) =>
                 <T>(source: Observable<ItemUpdate<T>>) =>
                     observe<T>(({ next, complete, error }) => {
                         const subscription = source
                             .filter(([, id]) => id === itemId)
                             .subscribe({
                                 next: ([item]) => item !== null && item !== undefined ? next(item) : complete(),
                                 complete,
                                 error
                             });
                         return () => subscription.unsubscribe();
                     });
             export const get = <T>(store: IndexedStore<T>) => {
                 const changes = hasNotifications<T>(store) ?
                     observe<ItemUpdate<T>>(({ next }) => {
                         const handle = after(store, "notify", (...args: ItemUpdate<T>) => next(args), true);
                         return () => handle.remove();
                     }) : empty;
                 return (id: string | number, opts: GetOpts = {}) =>
                     observe<T>(({ next, complete, error }) => {
                         try {
                             const result = store.get(id);
                             const handle = (x: T | null | undefined) => {
                                 if (x !== null && x !== undefined)
                                     next(x);
                                 complete();
                             };
                             if (isPromise(result)) {
                                 result.then(handle).then(undefined, error);
                                 if (isCancellable(result))
                                     return () => result.cancel();
                             } else {
                                 handle(result);
                             }
                         } catch (e) {
                             error(e);
                         }
                     }).cat(opts.observe !== false ? changes.pipe(filterItem(id)) : empty);
             };

djx/src/test/ts/observable-tests.ts +18 0

             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);
                 complete();
                 next(2);
             });
             const consumer1 = {
                 sum: 0,
                 next(v: number) {
                     this.sum += v;
                 }
             };
             subj1.subscribe(consumer1);
             tap.equal(consumer1.sum, 1, "Should get only one value");
             subj1.subscribe(consumer1);
             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 => {
                     tap.comment(`map1: ${v * 2}`);
                     maps++;
                     return v * 2;
                 })
                 .map(v => {
                     tap.comment(`map2: ${v * 2}`);
                     maps++;
                     return v * 2;
                 })
                 .map(v => {
                     tap.comment(`map3: ${v * 2}`);
                     maps++;
                     return v * 2;
                 })
                 .subscribe(consumer2);
             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 () => {
                     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 => observe(({ 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);
             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 => tap.comment("subj2: reduce =", val),
                 complete: () => tap.comment("subj2: complete")
             });
             tap.test("of(...) tests", async t => {
                 await subj2.reduce((a, b) => a + b).next()
                     .then(value => t.comment("subj2: next reduce=", value));
                 await subj2.next().then(val => t.equal(val, 1, "Should peek the first element"));
                 const cancelled = new Cancellation(cancel => cancel());
                 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
                 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(() => { });
             tap.test(".tap() tests", async t => {
                 const side: number[] = [];
                 of(1, 2)
                     .tap({ next: v => side.push(v), complete: () => side.push(0) })
                     .tap({ next: v => side.push(v * v) })
                     .subscribe({});
                 t.same(side, [1, 1, 2, 4, 0], ".tap() should be called in the order of registration");
                 side.length = 0;
                 await new Promise<void>(resolve => {
                     of(1, 2, delay(1).then(() => 3))
                         .tap({ next: v => side.push(v) })
                         .tap({ next: v => v === 1 && resolve() })
                         .subscribe({});
                 });
                 t.same(side, [1, 2], ".tap() should be processed synchronously");
+            }).catch(() => { });
+            tap.test(".while() tests", async t => {
+                const seq = of(1, 2, 3, 4).while(v => v <= 2);
+                t.same(await seq.collect(), [1, 2], "Should collect only taken elements");
+                const data: number[] = [];
+                let complete = 0;
+                seq.subscribe({
+                    next: v => data.push(v),
+                    complete: () => complete++
+                });
+                t.same(data, [1, 2], "Should receive only taken elements");
+                t.equal(complete, 1, "Complete should run once");
             }).catch(() => { });

playground/build.gradle +2 -2

             plugins {
                 id "org.implab.gradle-typescript" version "1.3.4"
                 id "ivy-publish"
             }
             def container = "djx-playground"
             configurations {
                 npmLocal
             }
             dependencies {
                 npmLocal project(":djx")
             }
             def bundleDir = fileTree(layout.buildDirectory.dir("bundle")) {
                 builtBy "bundle"
             }
             typescript {
                 compilerOptions {
                     lib = ["es5", "dom", "scripthost", "es2015.promise", "es2015.symbol", "es2015.iterable"]
                     // listFiles = true
                     strict = true
                     types = ["requirejs", "@implab/dojo-typings", "@implab/djx"]
                     module = "amd"
                     it.target = "es5"
                     experimentalDecorators = true
                     noUnusedLocals = false
                     jsx = "react"
                     jsxFactory = "createElement"
                     moduleResolution = "node"
                     // dojo-typings are sick
                     skipLibCheck = true
                     // traceResolution = true
                     // baseUrl = "./"
                     // paths = [ "*": [ "$projectDir/src/typings/*" ] ]
                     // baseUrl = "$projectDir/src/typings"
                     // typeRoots = ["$projectDir/src/typings"]
                 }
                 tscCmd = "$projectDir/node_modules/.bin/tsc"
                 tsLintCmd = "$projectDir/node_modules/.bin/tslint"
             	esLintCmd = "$projectDir/node_modules/.bin/eslint"
             }
             tasks.matching{ it.name =~ /^configureTs/ }.configureEach {
                 compilerOptions {
                     if (symbols != 'none') {
                         sourceMap = true
                         switch(symbols) {
                             case "local":
                                 sourceRoot = ( isWindows ? "file:///" : "file://" ) + it.rootDir
                             break;
                         }
                     }
                 }
             }
-            npmInstall {
+            task npmInstallLocalDeps {
-                //npmInstall.dependsOn it
+                npmInstall.dependsOn it
                 dependsOn configurations.npmLocal
                 doFirst {
                     configurations.npmLocal.each { f ->
                         exec {
                             commandLine "npm", "install", f, "--save-dev"
                         }
                     }
                 }
             }
             clean {
                 doFirst {
                     delete "$buildDir/bundle"
                 }
             }
             task processResourcesBundle(type: Copy) {
                 from "src/bundle"
                 into layout.buildDirectory.dir("bundle")
             }
             task copyModules(type: Copy) {
                 dependsOn npmInstall
                 into layout.buildDirectory.dir("bundle/js");
                 def pack = { String jsmod ->
                     into(jsmod) {
                         from npm.module(jsmod)
                     }
                 }
                 pack("@implab/djx")
                 pack("@implab/core-amd")
                 into("@js-joda/core") {
                     from(npm.module("@js-joda/core/dist"))
                 }
                 pack("dojo")
                 pack("dijit")
                 into("rxjs") {
                     from(npm.module("rxjs/dist/bundles"))
                 }
                 from npm.module("requirejs/require.js")
             }
             npmPublish {
                 enabled = false
             }
             task copyApp(type: Copy) {
                 dependsOn assemble
                 from typescript.assemblyDir
                 into layout.buildDirectory.dir("bundle/js/app")
             }
             task bundle {
                 dependsOn copyModules, processResourcesBundle, copyApp
             }
             task up(type: Exec) {
                 dependsOn bundle
                 commandLine "podman", "run", "--rm", "-d",
                     "--name", container,
                     "-p", "2078:80",
                     "-v", "$buildDir/bundle:/srv/www/htdocs",
                     "registry.implab.org/implab/apache2:latest"
             }
             task stop(type: Exec) {
                 commandLine "podman", "stop", container
             }

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