implab/implabjs-djx Commit - r142:894b8239b953

code review, minor refactoring

cin -

r142:894b8239b953 v1.9.0-rc5 default

parent child

Context file:

r142:894b8239b953

Collapse all files

djx/src/main/ts/observable.ts +32 -24

                 /**
                  * Called for the next element in the sequence
                  */
-                next?: (value: T) => void;
+                next?(value: T): void;
                 /**
                  * Called once when the error occurs in the sequence.
                  */
-                error?: (e: unknown) => void;
+                error?(e: unknown): void;
                 /**
                  * Called once at the end of the sequence.
                  */
-                complete?: () => void;
+                complete?(): void;
             }
             /**
             export type Producer<T> = (sink: Sink<T>) => (void | (() => void));
+            type FusedSink<T> = Omit<Sink<T>, "isClosed">;
+            type FusedProducer<T> = (sink: FusedSink<T>) => (void | (() => void));
             export interface Unsubscribable {
                 unsubscribe(): void;
             }
             const errorFallback = (e: unknown) => trace.error("Unhandled observable error: {0}", e);
             const sink = <T>(consumer: Observer<T>) => {
+                // eslint-disable-next-line @typescript-eslint/unbound-method
                 const { next, error, complete } = consumer;
                 return {
                     next: next ? next.bind(consumer) : noop,
-                    error: error ? error.bind(consumer) : errorFallback,
+                    error: error ? error.bind(consumer) : errorFallback, // report unhandled errors
-                    complete: complete ? complete.bind(consumer) : noop,
+                    complete: complete ? complete.bind(consumer) : noop
-                    isClosed: () => false
                 };
             };
             /** Wraps the producer to handle tear down logic and subscription management
              *
+             * The resulting producer will invoke cleanup logic on error or complete events
+             * and will prevent calling of any method from the sink.
+             *
              * @param producer The producer to wrap
              * @returns The wrapper producer
              */
-            const fuse = <T>(producer: Producer<T>) => ({ next, error, complete }: Sink<T>) => {
+            const fuse = <T>(producer: Producer<T>) => ({ next, error, complete }: FusedSink<T>) => {
                 let done = false;
                 let cleanup = noop;
                     complete: _fin(complete),
                     isClosed: () => done
                 };
+                // call the producer
                 cleanup = producer(safeSink) ?? noop;
+                // 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
                 return done ? cleanup() : _fin0;
             };
-            const _observe = <T>(producer: Producer<T>): Observable<T> => ({
+            const _observe = <T>(producer: FusedProducer<T>): Observable<T> => ({
                 subscribe: (consumer: Observer<T>) => ({
                     unsubscribe: producer(sink(consumer)) ?? noop
                 }),
                     }
                 }),
-                reduce: <A>(...args: [AccumulatorFn<T, A>, A] | [AccumulatorFn<T, T>]) => _observe<T | A>(({ next, complete, error, ...rest }) => {
+                reduce: <A>(...args: [AccumulatorFn<T, A>, A] | [AccumulatorFn<T, T>]) => _observe<T | A>(({ next, complete, error }) => {
                     if (args.length === 1) {
                         const [accumulator] = args;
                         let _acc: T;
                                     complete();
                                 }
                             },
-                            error,
+                            error
-                            ...rest
                         });
                     } else {
                         const [accumulator, initial] = args;
                                 next(_acc);
                                 complete();
                             },
-                            error,
+                            error
-                            ...rest
                         });
                     }
                 }),
                 next: collect(
                     producer,
-                    ({ next, complete, error, isClosed }) => ({
+                    ({ next, complete, error }) => ({
                         next: v => (next(v), complete()),
                         complete: () => error(new Error("The sequence is empty")),
-                        error,
+                        error
-                        isClosed
                     })
                 ),
                 collect: collect(
                     producer,
-                    ({ next, complete, ...rest }) => {
+                    ({ next, complete, error}) => {
                         const data: T[] = [];
                         return {
                             next: v => data.push(v),
                             complete: () => (next(data), complete()),
-                            ...rest
+                            error
                         };
                     }
                 )
             });
             const collect = <T, U>(
-                producer: Producer<T>,
+                producer: FusedProducer<T>,
-                collector: (result: Sink<U>) => Sink<T>
+                collector: (result: FusedSink<U>) => FusedSink<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 :
+                        producer(collector({ next, complete, error })) ?? noop :
                         noop;
                     return () => {
                 fused({
                     next: resolve,
                     error: reject,
-                    complete: noop,
+                    complete: noop
-                    isClosed: () => false
                 });
             });
             /** Converts a subscribable to the observable */
             export const ofSubscribable = <T>(subscribable: Subscribable<T>) =>
-                observe(sink => {
+                observe<T>(sink => {
                     const subscription = subscribable.subscribe(sink);
                     return () => subscription.unsubscribe();
                 });

djx/src/main/ts/operators/subject.ts +18 -8

             const noop = () => { };
-            /** Joins multiple subscribers to the single one. The resulting subscriber
+            /** Connects multiple subscribers to the single producer. The producer
              * will be created when the first client subscribes and will be released
              * with the the last client unsubscribed.
              *
              * Use this wrapper to prevent spawning multiple producers.
              *
+             * The emitted values are not cached therefore the new subscriber will not receive
+             * the values emitted before it has been subscribed.
+             *
              * @param source The source observable
-             * @returns The wrapped producer
+             * @returns The new observable
              */
             export const subject = <T>(source: Subscribable<T>): Observable<T> => {
-                let subscribers: Sink<T>[] = [];
+                let subscribers: Sink<T>[] = []; // the list of active subscribers
-                let subscription = { unsubscribe: noop };
+                let subscription = { unsubscribe: noop }; // current subscription
                 // cleanup method to release resources held by this subscription
                 const cleanup = (cb: (item: Sink<T>) => void) => {
                     const _subscribers = subscribers;
-                    subscribers = [];
+                    subscribers = []; // this will prevent a client cleanup login to run
                     _subscribers.forEach(cb);
-                    subscription.unsubscribe();
+                    // we don't need subscription.unsubscribe(), because cleanup is called
+                    // from complete or error methods.
                 };
                 const sink: Sink<T> = {
                 return observe(client => {
                     const _subscribers = subscribers;
                     subscribers.push(client);
-                    if (subscribers.length === 1)
+                    if (subscribers.length === 1) // this is the first client
-                        subscription = source.subscribe(sink);
+                        subscription = source.subscribe(sink); // activate the producer
                     return () => {
+                        // this is a cleanup logic for an individual client
                         if (_subscribers === subscribers) {
+                            // is the current subscription to the producer is active
+                            // find this client in the list of subscribers
                             const pos = subscribers.indexOf(client);
                             if (pos >= 0)
                                 subscribers.splice(pos, 1);
+                            // is this is the last subscriber we need to release the producer
                             if (!subscribers.length)
                                 subscription.unsubscribe();
                         }

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