fix: worker kept open when it shouldn't

projects/observable-webworker/src/lib/run-worker.spec.ts

@@ -1,5 +1,6 @@
 import { fakeAsync, tick } from '@angular/core/testing';
 import { BehaviorSubject, Notification, Observable, of } from 'rxjs';
+import { map } from 'rxjs/operators';
 import {
   DoTransferableWork,
   DoTransferableWorkUnit,
@@ -137,6 +138,50 @@ describe('runWorker', () => {
     sub.unsubscribe();
   });
 
+  // https://github.com/cloudnc/observable-webworker/issues/116
+  it('should complete the notification stream when the worker completes', () => {
+    const postMessageSpy = spyOn(window, 'postMessage');
+    postMessageSpy.calls.reset();
+
+    class TestWorker implements DoWork<number, number> {
+      public work(input$: Observable<number>): Observable<number> {
+        // here nothing should keep the subscription alive when input$ completes
+        return input$.pipe(map(input => input * 2));
+      }
+    }
+
+    const sub = runWorker(TestWorker);
+
+    const notificationEvent: WorkerMessageNotification<number> = new MessageEvent('message', {
+      data: new Notification('N', 10),
+    });
+
+    self.dispatchEvent(notificationEvent);
+
+    expect(postMessageSpy).toHaveBeenCalledWith(
+      jasmine.objectContaining({
+        kind: 'N',
+        value: 20,
+      }),
+    );
+
+    const completeEvent: WorkerMessageNotification<number> = new MessageEvent('message', {
+      data: new Notification('C'),
+    });
+
+    self.dispatchEvent(completeEvent);
+
+    expect(postMessageSpy).toHaveBeenCalledWith(
+      jasmine.objectContaining({
+        kind: 'C',
+      }),
+    );
+
+    // do note here that instead of manually closing the subscription
+    // we check it's already closed as expected
+    expect(sub.closed).toBeTrue();
+  });
+
   it('should not complete the notification stream if the worker does not complete', () => {
     const postMessageSpy = spyOn(window, 'postMessage');
     postMessageSpy.calls.reset();

projects/observable-webworker/src/lib/run-worker.ts

@@ -1,5 +1,5 @@
 import { from, fromEvent, Notification, Observable, Subscription } from 'rxjs';
-import { concatMap, dematerialize, filter, map, materialize } from 'rxjs/operators';
+import { concatMap, dematerialize, map, materialize } from 'rxjs/operators';
 import { DoTransferableWork, DoWork, DoWorkUnit, WorkerMessageNotification } from './observable-worker.types';
 
 export type ObservableWorkerConstructor<I = any, O = any> = new (...args: any[]) => DoWork<I, O> | DoWorkUnit<I, O>;
@@ -27,8 +27,6 @@ export function getWorkerResult<I, O>(
   const input$ = incomingMessages$.pipe(
     map((e: WorkerMessageNotification<I>): Notification<I> => e.data),
     map((n: Notification<I>) => new Notification(n.kind, n.value, n.error)),
-    // ignore complete, the calling thread will manage termination of the stream
-    filter(n => n.kind !== 'C'),
     dematerialize(),
   );