[WIP] Improve observable reuse and cleanup

reactfire/auth/auth.test.tsx

@@ -3,7 +3,7 @@ import { auth, User } from 'firebase/app';
 import '@testing-library/jest-dom/extend-expect';
 import * as React from 'react';
 import { AuthCheck, useUser } from '.';
-import { FirebaseAppProvider } from '..';
+import { FirebaseAppProvider, clearCache } from '..';
 import { Observable, Observer } from 'rxjs';
 import { act } from 'react-dom/test-utils';
 
@@ -62,6 +62,7 @@ describe('AuthCheck', () => {
 
   afterEach(() => {
     cleanup();
+    clearCache();
     jest.clearAllMocks();
   });
 
@@ -115,6 +116,7 @@ describe('AuthCheck', () => {
     await wait(() => expect(getByTestId('signed-out')).toBeInTheDocument());
   });
 
+  // implement this once we have an auth emulator
   test.todo('checks requiredClaims');
 });
 

reactfire/auth/index.tsx

@@ -12,11 +12,10 @@ import { from } from 'rxjs';
 
 export function preloadUser(firebaseApp: firebase.app.App) {
   return preloadAuth(firebaseApp).then(auth => {
-    const result = preloadObservable(
+    return preloadObservable(
       user(auth() as firebase.auth.Auth),
       'auth: user'
-    );
-    return result.request.promise;
+    ).getPromise();
   });
 }
 

reactfire/firebaseApp/sdk.tsx

@@ -1,4 +1,4 @@
-import { useFirebaseApp, preloadRequest, usePreloadedRequest } from '..';
+import { useFirebaseApp, preloadPromise, usePreloadedObservable } from '..';
 enum SDK {
   ANALYTICS = 'analytics',
   AUTH = 'auth',
@@ -74,13 +74,17 @@ function fetchSDK(sdk: SDK, firebaseApp: firebase.app.App) {
 function useSDK(sdk: SDK, firebaseApp?: firebase.app.App) {
   firebaseApp = firebaseApp || useFirebaseApp();
 
+  // before we worry about fetching, check to see if the SDK is already loaded
+  if (firebaseApp[sdk]) {
+    return firebaseApp[sdk];
+  }
+
+  const requestId = `firebase-sdk-${sdk}`;
+
   // use the request cache so we don't issue multiple fetches for the sdk
-  const result = preloadRequest(
-    () => fetchSDK(sdk, firebaseApp),
-    `firebase-sdk-${sdk}`
-  );
+  preloadPromise(() => fetchSDK(sdk, firebaseApp), requestId);
 
-  return usePreloadedRequest(result);
+  return usePreloadedObservable(requestId);
 }
 
 export function preloadAuth(firebaseApp: firebase.app.App) {

reactfire/firestore/index.tsx

@@ -28,7 +28,7 @@ export function preloadFirestoreDoc(
 ) {
   return preloadFirestore(firebaseApp).then(firestore => {
     const ref = refProvider(firestore() as firebase.firestore.Firestore);
-    return preloadObservable(doc(ref), ref.path);
+    return preloadObservable(doc(ref), 'firestore doc: ' + ref.path);
   });
 }
 
@@ -110,7 +110,7 @@ export function useFirestoreCollectionData<T = { [key: string]: unknown }>(
   query: firestore.Query,
   options?: ReactFireOptions<T[]>
 ): T[] {
-  const queryId = getHashFromFirestoreQuery(query);
+  const queryId = getHashFromFirestoreQuery(query) + ' data only';
 
   return useObservable(
     collectionData(query, checkIdField(options)),

reactfire/package.json

@@ -9,7 +9,7 @@
     "build-dev": "tsc --watch",
     "test-dev": "jest --verbose --watch",
     "emulators": "firebase emulators:start --only firestore,database",
-    "test": "firebase emulators:exec --only firestore,database \"jest --no-cache --verbose --detectOpenHandles --forceExit\"",
+    "test": "firebase emulators:exec --only firestore,database \"jest --no-cache --detectOpenHandles --forceExit\"",
     "copy-package-json": "cp package.pub.json pub/reactfire/package.json",
     "watch": "yarn build && tsc --watch",
     "build": "rm -rf pub && tsc && yarn copy-package-json && cp ../README.md pub/reactfire/README.md && cp ../LICENSE pub/reactfire/LICENSE && rollup -c"

reactfire/performance/performance.test.tsx

@@ -4,7 +4,7 @@ import '@testing-library/jest-dom/extend-expect';
 import * as React from 'react';
 import { Subject } from 'rxjs';
 import { SuspenseWithPerf } from '.';
-import { FirebaseAppProvider, useObservable } from '..';
+import { FirebaseAppProvider, useObservable, clearCache } from '..';
 
 const traceStart = jest.fn();
 const traceEnd = jest.fn();
@@ -36,6 +36,7 @@ const Provider = ({ children }) => (
 describe('SuspenseWithPerf', () => {
   afterEach(() => {
     cleanup();
+    clearCache();
     jest.clearAllMocks();
   });
 
@@ -206,7 +207,7 @@ describe('SuspenseWithPerf', () => {
       return (
         <SuspenseWithPerf
           traceId={'hello'}
-          fallback={'loading'}
+          fallback={<h1 data-testid="fallback">Actual</h1>}
           firePerf={(mockPerf() as unknown) as performance.Performance}
         >
           <Comp />
@@ -215,7 +216,8 @@ describe('SuspenseWithPerf', () => {
     };
 
     // render SuspenseWithPerf and go through normal trace start -> trace stop
-    const { getByTestId, rerender } = render(<Component />);
+    const { getByTestId } = render(<Component />);
+    await waitForElement(() => getByTestId('fallback'));
     expect(createTrace).toHaveBeenCalledTimes(1);
     act(() => o$.next('some value'));
     await waitForElement(() => getByTestId('child'));

reactfire/storage/index.tsx

@@ -3,6 +3,7 @@ import { storage } from 'firebase/app';
 import { getDownloadURL } from 'rxfire/storage';
 import { Observable } from 'rxjs';
 import { ReactFireOptions, useObservable, useFirebaseApp } from '..';
+import { useStorage } from '../firebaseApp';
 
 /**
  * modified version of rxFire's _fromTask
@@ -75,7 +76,7 @@ export function StorageImage(
 ) {
   let { storage, storagePath, ...imgProps } = props;
 
-  storage = storage || useFirebaseApp().storage();
+  storage = storage || useStorage()();
 
   const imgSrc = useStorageDownloadURL(storage.ref(storagePath));
   return <img src={imgSrc} {...imgProps} />;

reactfire/useObservable/ActiveObservable.ts

@@ -0,0 +1,81 @@
+import { first, tap, takeUntil, shareReplay, startWith } from 'rxjs/operators';
+import { Observable, timer, observable } from 'rxjs';
+type Observer<T> = import('rxjs').Observer<T>;
+
+export class ActiveObservable {
+  observable$: Observable<any>;
+  isReady: boolean;
+  value: any = undefined;
+  error: Error;
+  subscribers: number = 0;
+
+  constructor(observable$: Observable<any>, startWithValue?) {
+    if (startWithValue) {
+      this.value = startWithValue;
+      this.isReady = true;
+      observable$ = observable$.pipe(startWith(startWithValue));
+    } else {
+      this.isReady = false;
+    }
+
+    // create a shared observable
+    // we need to keep track of the latest value emitted so that we can tell React
+    // whether the component subscribed to this observable is ready to render
+    this.observable$ = observable$.pipe(
+      tap(
+        newVal => {
+          this.setValue(newVal);
+        },
+        error => {
+          this.setError(error);
+          throw error;
+        }
+      ),
+      shareReplay({
+        bufferSize: 1, // only cache the latest value
+        refCount: true // clean up the subscription if we don't have any subscribers
+      })
+    );
+  }
+
+  subscribeTemporarily(timeout: number, onComplete) {
+    this.subscribers++;
+
+    this.observable$
+      .pipe(takeUntil(timer(timeout)))
+      .subscribe(
+        () => {},
+        () => {},
+        () => {
+          this.subscribers--;
+          // onComplete()
+        }
+      )
+      .add(onComplete);
+  }
+
+  subscribe(observer: Observer<any>) {
+    this.subscribers++;
+    const subscription = this.observable$.subscribe(observer);
+
+    subscription.add(() => {
+      console.log('UNSUBSCRIBE CALLED');
+      this.subscribers--;
+    });
+    return subscription;
+  }
+
+  getPromise() {
+    return this.observable$.pipe(first()).toPromise();
+  }
+
+  setValue = value => {
+    this.value = value;
+    this.isReady = true;
+  };
+
+  setError = err => {
+    this.error = err;
+    this.isReady = true;
+  };
+}

reactfire/useObservable/index.ts

@@ -1,48 +1,61 @@
 import * as React from 'react';
-import { Observable } from 'rxjs';
-import { first, startWith } from 'rxjs/operators';
-import { ActiveRequest, ObservablePromiseCache } from './requestCache';
-
-const requestCache = new ObservablePromiseCache();
-
-export function preloadRequest(
-  getPromise,
-  requestId: string
-): { requestId: string; request: ActiveRequest } {
-  const request = requestCache.createDedupedRequest(getPromise, requestId);
-
-  return {
-    requestId: requestId,
-    request
-  };
+import { Observable, timer, from } from 'rxjs';
+import { first, startWith, shareReplay, takeUntil } from 'rxjs/operators';
+import { ActiveObservable } from './ActiveObservable';
+
+const PRELOAD_SUBSCRIBE_TIME = 5 /*seconds*/ * 1000;
+const observableCache = new Map<string, ActiveObservable>();
+
+export function preloadPromise(getPromise, requestId: string): Promise<any> {
+  const activeObservable = preloadObservable(from(getPromise()), requestId);
+
+  return activeObservable.getPromise();
 }
 
-// Starts listening to an Observable.
-// Call this once you know you're going to render a
-// child that will consume the observable
+// Registers an Observable and starts listening to it to prime it for when a real
+// subscriber starts listening
 export function preloadObservable(
   observable$: Observable<any>,
-  observableId: string
-): { requestId: string; request: ActiveRequest } {
-  return preloadRequest(
-    () => observable$.pipe(first()).toPromise(),
-    observableId
-  );
+  observableId: string,
+  startWithValue?
+): ActiveObservable {
+  // If we already have something in the cache with that ID, re-use it
+  if (observableCache.has(observableId)) {
+    return observableCache.get(observableId);
+  }
+
+  const activeObservable = new ActiveObservable(observable$, startWithValue);
+  observableCache.set(observableId, activeObservable);
+
+  // subscribe to the observable so that we can get a value
+  // this subscription cleans itself up after PRELOAD_SUBSCRIBE_TIME passes
+  activeObservable.subscribeTemporarily(PRELOAD_SUBSCRIBE_TIME, () => {
+    if (activeObservable.subscribers === 0) {
+      console.log(`CLEANING UP ${observableId}`);
+      observableCache.delete(observableId);
+    }
+  });
+
+  return activeObservable;
 }
 
-export function usePreloadedRequest(preloadResult: { requestId: string }) {
-  const request = requestCache.getRequest(preloadResult.requestId);
+export function usePreloadedObservable(observableId: string) {
+  if (!observableCache.has(observableId)) {
+    throw new Error(`Observable "${observableId}" doesn't exist!`);
+  }
+
+  const activeObservable = observableCache.get(observableId);
 
   // Suspend if we're not ready yet
-  if (!request.isComplete) {
-    throw request.promise;
+  if (!activeObservable.isReady) {
+    throw activeObservable.getPromise();
   }
 
-  if (request.error) {
-    throw request.error;
+  if (activeObservable.error) {
+    throw activeObservable.error;
   }
 
-  return request.value;
+  return activeObservable.value;
 }
 
 export function useObservable(
@@ -54,36 +67,52 @@ export function useObservable(
     throw new Error('cannot call useObservable without an observableId');
   }
 
-  const result = preloadObservable(observable$, observableId);
+  // register observable in the cache
+  const activeObservable = preloadObservable(
+    observable$,
+    observableId,
+    startWithValue
+  );
 
-  let initialValue = startWithValue;
-  if (initialValue === undefined) {
+  if (activeObservable.isReady === false) {
     // this will Suspend until the Promise resolves
-    initialValue = usePreloadedRequest(result);
+    usePreloadedObservable(observableId);
   }
 
-  const [latestValue, setValue] = React.useState(initialValue);
+  if (activeObservable.error) {
+    throw activeObservable.error;
+  }
 
-  React.useEffect(() => {
-    const subscription = observable$.pipe(startWith(initialValue)).subscribe(
-      newVal => {
-        // update the value in requestCache
-        result.request.setValue(newVal);
+  const [latestValue, setValue] = React.useState(activeObservable.value);
 
-        // update state
+  React.useEffect(() => {
+    const subscription = activeObservable.subscribe({
+      next: newVal => {
         setValue(newVal);
       },
-      error => {
-        console.error('There was an error', error);
+      error: error => {
         throw error;
-      }
-    );
+      },
+      complete: () => {}
+    });
 
     return () => {
       subscription.unsubscribe();
-      requestCache.removeRequest(observableId);
+
+      // if we were the last subscriber, remove the observable from the cache
+      if (activeObservable.subscribers === 0) {
+        observableCache.delete(observableId);
+      }
     };
-  }, [observableId]);
+  }, [activeObservable]);
 
   return latestValue;
 }
+
+export function clearCache() {
+  observableCache.clear();
+}
+
+export function getCache() {
+  return observableCache;
+}