Require passing onNext callback to observable.subscribe (#124)

Fix the `observable.subscribe()` type - it now requires an onNext callback.

This PR also fixes a leak in `useObservers` hook where onNext callbacks were not properly cleared when the hook was unmounted.

.vscode/settings.json

@@ -2,6 +2,7 @@
   "cSpell.words": [
     "Middlewares",
     "MVVM",
-    "unmagler"
+    "unmagler",
+    "unsubscribers"
   ]
 }

src/observable/Observable.ts

@@ -1,6 +1,5 @@
 import { Observable as IObservable, OnNext, Unsubscribe } from './types';
 
-const NOOP = () => {};
 export class Observable<T> implements IObservable<T> {
   private subscribers: Set<OnNext<T>> = new Set();
   private currentValue: T | undefined;
@@ -28,7 +27,7 @@ export class Observable<T> implements IObservable<T> {
     });
   }
 
-  public subscribe(onNext: OnNext<T> = NOOP): Unsubscribe {
+  public subscribe(onNext: OnNext<T>): Unsubscribe {
     if (this.subscribers.has(onNext)) {
       throw new Error('Subscriber already subscribed');
     }

src/observable/mediator/MediatorObservable.test.ts

@@ -64,7 +64,6 @@ describe('MediatorObservable', () => {
       uut.value *= nextC;
     });
 
-    uut.subscribe(() => {});
     a.value = 1;
     expect(uut.value).toEqual(1);
     b.value = 2;
@@ -102,7 +101,6 @@ describe('MediatorObservable', () => {
       uut.value = parseInt(nextB, 10);
     });
 
-    uut.subscribe(() => {});
     a.value = 1;
     expect(uut.value).toEqual(1);
     b.value = '2';

src/observable/useObservers.test.ts

@@ -1,4 +1,5 @@
 import { act, renderHook } from '@testing-library/react';
+import _ from 'lodash';
 import { Observable } from './Observable';
 import { useObservers } from './useObservers';
 
@@ -31,4 +32,13 @@ describe('useObservers', () => {
     act(() => { fooObservable.value = 1; });
     expect(result.current.foo).toBe(1);
   });
+
+  it('should release onNext callbacks when unmounted', () => {
+    const { unmount } = renderHook(uut);
+    unmount();
+
+    [fooObservable, barObservable, bazObservable].forEach((observable) => {
+      expect(_.get(observable, 'subscribers.size')).toBe(0);
+    });
+  });
 });

src/observable/useObservers.ts

@@ -1,20 +1,25 @@
 import { useEffect, useState } from 'react';
 import { MediatorObservable } from './mediator/MediatorObservable';
-import { ObservedValues } from './types';
+import { ObservedValues, Unsubscribe } from './types';
 import { mapObservablesToValues } from './mapObservablesToValues';
 
 export function useObservers<T extends Record<string, any>>(observables: T): ObservedValues<T> {
   const [values, setValues] = useState(() => mapObservablesToValues(observables));
 
   useEffect(() => {
     const mediator = new MediatorObservable();
+    const unsubscribers: Unsubscribe[] = [];
+
     Object.keys(observables as {}).forEach((key) => {
-      mediator.addSource(observables[key], (value) => {
-        setValues({ ...values, [key]: value });
+      const onNext = (value: any) => setValues({ ...values, [key]: value });
+      mediator.addSource(observables[key], onNext);
+
+      unsubscribers.push(() => {
+        observables[key].unsubscribe(onNext);
       });
     });
 
-    return mediator.subscribe();
+    return () => unsubscribers.forEach((unsubscribe) => unsubscribe());
   }, []);
 
   return values;