implab/implabjs-djx Files · djx/src/main/ts/observable.ts

bump dependencies, bump ts to 5.2...

bump dependencies, bump ts to 5.2 moved observables logic to ObservableImpl fixed while/until bug with complete handler multiple call

cin - - Load All Authors

File last commit:

r155:6acbe6efbe20 v1.10.2 default


                r155:6acbe6efbe20

v1.10.2 default

Download file

             observable.ts
        
                    104 lines
            
             | 3.6 KiB
            
                | video/mp2t
            
             |
                TypeScriptLexer
            
             / djx / src / main / ts / observable.ts
          
                    History
                
                 |
                  Source
                 | Raw
                 |Copy content
                 |Copy permalink

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

              r129
            
      import { isPromise } from "@implab/core-amd/safe";

        cin
    
bump dependencies, bump ts to 5.2...

              r155
            
      import { Unsubscribable, Producer, FusedSink, Observable, Subscribable } from "./observable/interfaces";

      import { ObservableImpl } from "./observable/ObservableImpl";

        cin
    
`Subscribable` is made compatible with rxjs, added map, filter and scan...

              r102
            
        cin
    
added whenRendered() method to wait for pending oprations to complete

              r118
            
      export const isUnsubscribable = (v: unknown): v is Unsubscribable =>

        cin
    
`Subscribable` is made compatible with rxjs, added map, filter and scan...

              r102
            
          v !== null && v !== undefined && typeof (v as Unsubscribable).unsubscribe === "function";

        cin
    
added whenRendered() method to wait for pending oprations to complete

              r118
            
      export const isSubscribable = <T = unknown>(v: unknown): v is Subscribable<T> =>

        cin
    
`Subscribable` is made compatible with rxjs, added map, filter and scan...

              r102
            
          v !== null && v !== undefined && typeof (v as Subscribable<unknown>).subscribe === "function";

        cin
    
refactoring, adding scope to rendering methods

              r96
            
        cin
    
`Subscribable` is made compatible with rxjs, added map, filter and scan...

              r102
            
      const noop = () => { };

        cin
    
corrected tear down logic handling in observables. Added support for observable query results

              r110
            
      /** Wraps the producer to handle tear down logic and subscription management

       * 

        cin
    
code review, minor refactoring

              r142
            
       * The resulting producer will invoke cleanup logic on error or complete events

       * and will prevent calling of any method from the sink. 

       * 

        cin
    
corrected tear down logic handling in observables. Added support for observable query results

              r110
            
       * @param producer The producer to wrap

       * @returns The wrapper producer

       */

        cin
    
code review, minor refactoring

              r142
            
      const fuse = <T>(producer: Producer<T>) => ({ next, error, complete }: FusedSink<T>) => {

        cin
    
`Subscribable` is made compatible with rxjs, added map, filter and scan...

              r102
            
          let done = false;

        cin
    
corrected tear down logic handling in observables. Added support for observable query results

              r110
            
          let cleanup = noop;

          const _fin = <A extends unknown[]>(fn: (...args: A) => void) =>

              (...args: A) => done ?

                  void (0) :

                  (done = true, cleanup(), fn(...args));

        cin
    
added while, until methods to the observable interface....

              r124
            
          const _fin0 = () => done ? void (0) : (done = true, cleanup());

        cin
    
corrected tear down logic handling in observables. Added support for observable query results

              r110
            
          const safeSink = {

        cin
    
linting

              r109
            
              next: (value: T) => { !done && next(value); },

        cin
    
corrected tear down logic handling in observables. Added support for observable query results

              r110
            
              error: _fin(error),

              complete: _fin(complete),

              isClosed: () => done

        cin
    
linting

              r109
            
          };

        cin
    
code review, minor refactoring

              r142
            
          // call the producer

        cin
    
corrected tear down logic handling in observables. Added support for observable query results

              r110
            
          cleanup = producer(safeSink) ?? noop;

        cin
    
code review, minor refactoring

              r142
            
          // if the producer throws exception bypass it to the caller rather then to 

          // the sink. This is a feature.

          // if the producer completed the sequence immediately call the cleanup in place

        cin
    
added while, until methods to the observable interface....

              r124
            
          return done ? cleanup() : _fin0;

        cin
    
linting

              r109
            
      };

        cin
    
refactoring, adding scope to rendering methods

              r96
            
        cin
    
bump dependencies, bump ts to 5.2...

              r155
            
      export const observe = <T>(producer: Producer<T>): Observable<T> => new ObservableImpl<T>(fuse(producer));

        cin
    
corrected tear down logic handling in observables. Added support for observable query results

              r110
            
        cin
    
added 'buffer' and 'subject' observable operators

              r133
            
      /** Converts an array to the observable sequence of its elements. */

        cin
    
bump dependencies, bump ts to 5.2...

              r155
            
      export const ofArray = <T>(items: T[]) => new ObservableImpl<T>(

        cin
    
added reduce() and next() methods to observable...

              r116
            
          ({ next, complete }) => (

              items.forEach(next),

              complete()

          )

      );

        cin
    
corrected tear down logic handling in observables. Added support for observable query results

              r110
            
        cin
    
added 'buffer' and 'subject' observable operators

              r133
            
      /** Converts a subscribable to the observable */

      export const ofSubscribable = <T>(subscribable: Subscribable<T>) =>

        cin
    
code review, minor refactoring

              r142
            
          observe<T>(sink => {

        cin
    
added 'buffer' and 'subject' observable operators

              r133
            
              const subscription = subscribable.subscribe(sink);

              return () => subscription.unsubscribe();

          });

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

              r129
            
      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())

        cin
    
added reduce() and next() methods to observable...

              r116
            
      );

        cin
    
corrected tear down logic handling in observables. Added support for observable query results

              r110
            
        cin
    
added 'buffer' and 'subject' observable operators

              r133
            
      /** Converts a list of parameter values to the observable sequence. The

       * order of elements in the list will be preserved in the resulting sequence.

       */

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

              r129
            
      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);

              }

          );

        cin
    
corrected tear down logic handling in observables. Added support for observable query results

              r110
            
        cin
    
bump dependencies, bump ts to 5.2...

              r155
            
      export const empty = new ObservableImpl<never>(({ complete }) => complete());

      export type * from "./observable/interfaces";

	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

cin added observable.collect() method to collect a sequnce to the array...	r129	import { isPromise } from "@implab/core-amd/safe";
cin bump dependencies, bump ts to 5.2...	r155	import { Unsubscribable, Producer, FusedSink, Observable, Subscribable } from "./observable/interfaces";
		import { ObservableImpl } from "./observable/ObservableImpl";
cin `Subscribable` is made compatible with rxjs, added map, filter and scan...	r102
cin added whenRendered() method to wait for pending oprations to complete	r118	export const isUnsubscribable = (v: unknown): v is Unsubscribable =>
cin `Subscribable` is made compatible with rxjs, added map, filter and scan...	r102	v !== null && v !== undefined && typeof (v as Unsubscribable).unsubscribe === "function";

cin added whenRendered() method to wait for pending oprations to complete	r118	export const isSubscribable = <T = unknown>(v: unknown): v is Subscribable<T> =>
cin `Subscribable` is made compatible with rxjs, added map, filter and scan...	r102	v !== null && v !== undefined && typeof (v as Subscribable<unknown>).subscribe === "function";
cin refactoring, adding scope to rendering methods	r96
cin `Subscribable` is made compatible with rxjs, added map, filter and scan...	r102	const noop = () => { };

cin corrected tear down logic handling in observables. Added support for observable query results	r110	/** Wraps the producer to handle tear down logic and subscription management
		*
cin code review, minor refactoring	r142	* The resulting producer will invoke cleanup logic on error or complete events
		* and will prevent calling of any method from the sink.
		*
cin corrected tear down logic handling in observables. Added support for observable query results	r110	* @param producer The producer to wrap
		* @returns The wrapper producer
		*/
cin code review, minor refactoring	r142	const fuse = <T>(producer: Producer<T>) => ({ next, error, complete }: FusedSink<T>) => {
cin `Subscribable` is made compatible with rxjs, added map, filter and scan...	r102	let done = false;
cin corrected tear down logic handling in observables. Added support for observable query results	r110	let cleanup = noop;

		const _fin = <A extends unknown[]>(fn: (...args: A) => void) =>
		(...args: A) => done ?
		void (0) :
		(done = true, cleanup(), fn(...args));

cin added while, until methods to the observable interface....	r124	const _fin0 = () => done ? void (0) : (done = true, cleanup());

cin corrected tear down logic handling in observables. Added support for observable query results	r110	const safeSink = {
cin linting	r109	next: (value: T) => { !done && next(value); },
cin corrected tear down logic handling in observables. Added support for observable query results	r110	error: _fin(error),
		complete: _fin(complete),
		isClosed: () => done
cin linting	r109	};
cin code review, minor refactoring	r142	// call the producer
cin corrected tear down logic handling in observables. Added support for observable query results	r110	cleanup = producer(safeSink) ?? noop;
cin code review, minor refactoring	r142	// if the producer throws exception bypass it to the caller rather then to
		// the sink. This is a feature.

		// if the producer completed the sequence immediately call the cleanup in place
cin added while, until methods to the observable interface....	r124	return done ? cleanup() : _fin0;
cin linting	r109	};
cin refactoring, adding scope to rendering methods	r96
cin bump dependencies, bump ts to 5.2...	r155	export const observe = <T>(producer: Producer<T>): Observable<T> => new ObservableImpl<T>(fuse(producer));
cin corrected tear down logic handling in observables. Added support for observable query results	r110
cin added 'buffer' and 'subject' observable operators	r133	/** Converts an array to the observable sequence of its elements. */
cin bump dependencies, bump ts to 5.2...	r155	export const ofArray = <T>(items: T[]) => new ObservableImpl<T>(
cin added reduce() and next() methods to observable...	r116	({ next, complete }) => (
		items.forEach(next),
		complete()
		)
		);
cin corrected tear down logic handling in observables. Added support for observable query results	r110
cin added 'buffer' and 'subject' observable operators	r133	/** Converts a subscribable to the observable */
		export const ofSubscribable = <T>(subscribable: Subscribable<T>) =>
cin code review, minor refactoring	r142	observe<T>(sink => {
cin added 'buffer' and 'subject' observable operators	r133	const subscription = subscribable.subscribe(sink);
		return () => subscription.unsubscribe();
		});

cin added observable.collect() method to collect a sequnce to the array...	r129	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())
cin added reduce() and next() methods to observable...	r116	);
cin corrected tear down logic handling in observables. Added support for observable query results	r110
cin added 'buffer' and 'subject' observable operators	r133	/** Converts a list of parameter values to the observable sequence. The
		* order of elements in the list will be preserved in the resulting sequence.
		*/
cin added observable.collect() method to collect a sequnce to the array...	r129	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);
		}
		);
cin corrected tear down logic handling in observables. Added support for observable query results	r110
cin bump dependencies, bump ts to 5.2...	r155	export const empty = new ObservableImpl<never>(({ complete }) => complete());

		export type * from "./observable/interfaces";