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

Implemented subscription SubscriptionImpl, fixed subscription resource management

cin - - Load All Authors

File last commit:

r158:078eca3dc271 v1.10.3 default


                r158:078eca3dc271

v1.10.3 default

Download file

             observable.ts
        
                    68 lines
            
             | 2.4 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
    
WIP observables

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

        cin
    
Implemented subscription SubscriptionImpl, fixed subscription resource management

              r158
            
      import { ObservableImpl, fuse } 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
    
Implemented subscription SubscriptionImpl, fixed subscription resource management

              r158
            
      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
    
WIP observables

              r157
            
      export const ofArray = <T>(items: T[]): Observable<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
    
WIP observables

              r157
            
      export const empty: Observable<never> = new ObservableImpl<never>(({ complete }) => complete());

        cin
    
bump dependencies, bump ts to 5.2...

              r155
            
      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 WIP observables	r157	import { Unsubscribable, Producer, Observable, Subscribable } from "./observable/interfaces";
cin Implemented subscription SubscriptionImpl, fixed subscription resource management	r158	import { ObservableImpl, fuse } 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 Implemented subscription SubscriptionImpl, fixed subscription resource management	r158	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 WIP observables	r157	export const ofArray = <T>(items: T[]): Observable<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 WIP observables	r157	export const empty: Observable<never> = new ObservableImpl<never>(({ complete }) => complete());
cin bump dependencies, bump ts to 5.2...	r155
		export type * from "./observable/interfaces";