Merge pull request #3 from ng-web-apis/bugfix/terminate

fix: close all subscriptions, if the worker was terminated

Author: IKatsuba

projects/workers/src/worker/classes/web-worker.spec.ts

@@ -89,4 +89,25 @@ describe('WebWorker', () => {
             'Uncaught reason',
         );
     });
+
+    it('should close all subscriptions, if the worker was terminated', async () => {
+        const worker = WebWorker.fromFunction<void, string>(() => 'some data');
+
+        const subscriptions = [
+            worker.subscribe(),
+            worker.subscribe(),
+            worker.subscribe(),
+        ];
+
+        worker.terminate();
+        expect(subscriptions.map(s => s.closed)).toEqual([true, true, true]);
+    });
+
+    it("shouldn't throw any errors, if the worker was terminated twice", async () => {
+        const worker = WebWorker.fromFunction<void, string>(() => 'some data');
+
+        worker.terminate();
+        worker.terminate();
+        expect(await worker.toPromise()).toBeUndefined();
+    });
 });

projects/workers/src/worker/classes/web-worker.ts

@@ -1,12 +1,13 @@
-import {EMPTY, fromEvent, merge, Observable} from 'rxjs';
-import {take, tap} from 'rxjs/operators';
+import {EMPTY, fromEvent, merge, Observable, Subject} from 'rxjs';
+import {take, takeUntil, tap} from 'rxjs/operators';
 import {WORKER_BLANK_FN} from '../consts/worker-fn-template';
 import {TypedMessageEvent} from '../types/typed-message-event';
 import {WorkerFunction} from '../types/worker-function';
 
 export class WebWorker<T = any, R = any> extends Observable<TypedMessageEvent<R>> {
     private readonly worker: Worker | undefined;
     private readonly url: string;
+    private readonly destroy$: Subject<void>;
 
     constructor(url: string, options?: WorkerOptions) {
         let worker: Worker | undefined;
@@ -19,26 +20,29 @@ export class WebWorker<T = any, R = any> extends Observable<TypedMessageEvent<R>
         }
 
         super(subscriber => {
+            let eventStream$: Observable<TypedMessageEvent<R> | ErrorEvent> = EMPTY;
+
             if (error) {
                 subscriber.error(error);
+            } else if (this.destroy$.isStopped) {
+                subscriber.complete();
+            } else if (worker) {
+                eventStream$ = merge(
+                    fromEvent<TypedMessageEvent<R>>(worker, 'message').pipe(
+                        tap(event => subscriber.next(event)),
+                    ),
+                    fromEvent<ErrorEvent>(worker, 'error').pipe(
+                        tap(event => subscriber.error(event)),
+                    ),
+                ).pipe(takeUntil(this.destroy$));
             }
 
-            const eventStream$ = worker
-                ? merge(
-                      fromEvent<TypedMessageEvent<R>>(worker, 'message').pipe(
-                          tap(event => subscriber.next(event)),
-                      ),
-                      fromEvent<ErrorEvent>(worker, 'error').pipe(
-                          tap(event => subscriber.error(event)),
-                      ),
-                  )
-                : EMPTY;
-
-            return eventStream$.subscribe();
+            eventStream$.subscribe().add(subscriber);
         });
 
         this.worker = worker;
         this.url = url;
+        this.destroy$ = new Subject<void>();
     }
 
     static fromFunction<T, R>(
@@ -57,7 +61,11 @@ export class WebWorker<T = any, R = any> extends Observable<TypedMessageEvent<R>
 
         worker.postMessage(data);
 
-        return promise;
+        return promise.then(result => {
+            worker.terminate();
+
+            return result;
+        });
     }
 
     private static createFnUrl(fn: WorkerFunction): string {
@@ -69,11 +77,18 @@ export class WebWorker<T = any, R = any> extends Observable<TypedMessageEvent<R>
     }
 
     terminate() {
+        if (this.destroy$.isStopped) {
+            return;
+        }
+
         if (this.worker) {
             this.worker.terminate();
         }
 
         URL.revokeObjectURL(this.url);
+
+        this.destroy$.next();
+        this.destroy$.complete();
     }
 
     postMessage(value: T) {

projects/workers/src/worker/pipes/worker.pipe.ts

@@ -12,7 +12,8 @@ export class WorkerPipe implements PipeTransform {
     private observers = new WeakMap<WebWorker, Observable<any>>();
 
     transform<T, R>(value: T, fn: WorkerFunction<T, R>): Observable<R> {
-        const worker = this.workers.get(fn) || WebWorker.fromFunction(fn);
+        const worker: WebWorker<T, R> =
+            this.workers.get(fn) || WebWorker.fromFunction(fn);
 
         this.workers.set(fn, worker);
         worker.postMessage(value);