implab/implabjs-djx Commit - r144:63215d91ae4b

added tap() method to observable...

cin -

r144:63215d91ae4b v1.9.0-rc6 default

parent child

Context file:

r144:63215d91ae4b

Collapse all files

djx/src/main/ts/observable.ts +20 -2

                  v !== null && v !== undefined && typeof (v as Subscribable<unknown>).subscribe === "function";
              export interface Subscribable<T> {
-                 subscribe(consumer: Observer<T>): Unsubscribable;
+                 /** Subscribes a consumer to events. If a consumer isn't specified
+                  * this method activates the producer to achieve side affects if any.
+                  */
+                 subscribe(consumer?: Observer<T>): Unsubscribable;
              }
              export type AccumulatorFn<T, A> = (acc: A, value: T) => A;
                   */
                  map<T2>(mapper: (value: T) => T2): Observable<T2>;
+                 /** Injects the specified observer into the each producer to consumer chain.
+                  * The method is used to add side effect to the events processing.
+                  *
+                  * @param observer The consumer for the events
+                  */
+                 tap(observer: Observer<T>): Observable<T>;
                  /** Filters elements of the sequence. The resulting sequence will
                   * contain only elements which match the specified predicate.
                   *
              };
              const _observe = <T>(producer: FusedProducer<T>): Observable<T> => ({
-                 subscribe: (consumer: Observer<T>) => ({
+                 subscribe: (consumer: Observer<T> = {}) => ({
                      unsubscribe: producer(sink(consumer)) ?? noop
                  }),
                      })
                  ),
+                 tap: ({next: tapNext, complete: tapComplete, error: tapError}) => _observe(({next,complete, error}) =>
+                     producer({
+                         next: tapNext ? (v => (tapNext(v), next(v))) : next,
+                         complete: tapComplete ? (() => (tapComplete(), complete())): complete,
+                         error: tapError ? (e => (tapError(e), error(e))) : error
+                     })
+                 ),
                  filter: (predicate) => _observe(({ next, ...rest }) =>
                      producer({
                          next: next !== noop ? (v: T) => predicate(v) ? next(v) : void (0) : noop,

djx/src/main/ts/store.ts +39 -12

              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 }) => {
+             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);
+                 };
+             };
+             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[] = [];
+                     let results: PromiseOrValue<T[]> = pending;
+                     const data = observe<OrderedUpdate<T>>(({ next, complete, error }) => {
                          const processResults = (items: T[]) =>
                              items.forEach((item, newIndex) => next({ item, newIndex, prevIndex: -1 }));
                          try {
-                             const results = store.query(query, options);
+                             if (results === pending)
+                                 results = store.query(query, options);
                              if (isPromise(results)) {
-                                 results.then(processResults).then(undefined, error);
+                                 results.then(processResults).then(complete, error);
+                                 if (isCancellable(results))
+                                     return results.cancel.bind(results);
                              } else {
                                  processResults(results);
+                                 complete();
                              }
+                         } catch (e) {
+                             error(e);
+                         }
+                     });
-                             if (!isClosed() && (options?.observe !== false) && isDjObservableResults<T>(results)) {
+                     const updates = observe<OrderedUpdate<T>>(({ next, complete, error, isClosed }) => {
+                         try {
+                             if (!isClosed() && isDjObservableResults<T>(results)) {
                                  const h = results.observe((item, prevIndex, newIndex) => next({ item, prevIndex, newIndex }), includeUpdates);
                                  return () => h.remove();
                              } else {
                                  complete();
                              }
-                         } catch (err) {
-                             error(err);
+                         } catch (e) {
+                             error(e);
                          }
                      });
+                     return [data, updates];
                  };
              interface IndexedStore<T> {
                  get(id: string | number): PromiseLike<T> | T | null | undefined;
              }
                  observe?: boolean;
              }
-             type Change<T> = [item: NonNullable<T>, id: string | number | undefined] |
+             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<Change<T>>) =>
+                 <T>(source: Observable<ItemUpdate<T>>) =>
                      observe<T>(({ next, complete, error }) => {
                          const subscription = source
                              .filter(([, id]) => id === itemId)
              export const get = <T>(store: IndexedStore<T>) => {
                  const changes = hasNotifications<T>(store) ?
-                     observe<Change<T>>(({ next }) => {
-                         const handle = after(store, "notify", (...args: Change<T>) => next(args), true);
+                     observe<ItemUpdate<T>>(({ next }) => {
+                         const handle = after(store, "notify", (...args: ItemUpdate<T>) => next(args), true);
                          return () => handle.remove();
                      }) : empty;

djx/src/test/ts/observable-store-tests.ts +25 -1

              import Memory = require("dojo/store/Memory");
              import Observerable = require("dojo/store/Observable");
-             import { get } from "./store";
+             import { get, queryEx } from "./store";
              import tap = require("tap");
              interface Person {
                      samSubscription.unsubscribe();
                  }
+                 store.put({ id: samId, name: "Sam", age: 29});
+                 const [data, updates] = queryEx(store)({ age: 29}, { sort: [{attribute: "id"}] });
+                 const dump: string[] = [];
+                 const subscription = data
+                     .tap({
+                         complete: () => dump.push("eof")
+                     })
+                     .cat(updates)
+                     .tap({
+                         next: ({item: {id}}) => dump.push(id),
+                         complete: () => dump.push("eof")
+                     })
+                     .subscribe({});
+                 t.same(dump, ["id:peter", "id:sam", "eof"]);
+                 store.put({ id: "id:mary", name: "Mary", age: 29});
+                 t.same(dump, ["id:peter", "id:sam", "eof", "id:mary"]);
+                 subscription.unsubscribe();
              }).catch(() => { });
  No newline at end of file

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

		@@ -147,4 +147,27 tap.test("of(...) tests", async t => {
147	147	t.same(await of(1,2,3).collect(), [1,2,3], ".collect() should return the collected sequence");
148	148	await t.rejects(of(1,2,3).collect(cancelled), ".collect() should support cancellation");
149	149
150		}).catch(() => { }); No newline at end of file
	150	}).catch(() => { });
	151
	152	tap.test(".tap() tests", async t => {
	153	const side: number[] = [];
	154
	155	of(1,2)
	156	.tap({next: v => side.push(v), complete: () => side.push(0)})
	157	.tap({next: v => side.push(v*v)})
	158	.subscribe({});
	159
	160	t.same(side, [1,1,2,4,0], ".tap() should be called in the order of registration");
	161
	162	side.length = 0;
	163
	164	await new Promise<void>(resolve => {
	165	of(1,2,delay(1).then(() => 3))
	166	.tap({next: v => side.push(v)})
	167	.tap({ next: v => v === 1 && resolve()})
	168	.subscribe({});
	169	});
	170
	171	t.same(side, [1,2], ".tap() should be processed synchronously");
	172
	173	}).catch(() => {}); No newline at end of file

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