fix(subscription): preserve observer this context

fix #122

packages/observable-hooks/__tests__/use-layout-subscription.spec.ts

@@ -1,7 +1,7 @@
-import { useLayoutSubscription } from "../src";
-import { renderHook, act } from "@testing-library/react-hooks";
-import { of, BehaviorSubject, Subject, throwError } from "rxjs";
+import { act, renderHook } from "@testing-library/react-hooks";
 import { useState } from "react";
+import { BehaviorSubject, Subject, of, throwError } from "rxjs";
+import { useLayoutSubscription } from "../src";
 import { mockConsoleError } from "./utils";
 
 describe("useLayoutSubscription", () => {
@@ -223,4 +223,25 @@ describe("useLayoutSubscription", () => {
     expect(numSpy).toBeCalledTimes(2);
     expect(numSpy).lastCalledWith(2);
   });
+
+  it("should subscribe with Subject", () => {
+    const num1$ = of(1);
+    const num2$ = new Subject();
+    const numSpy = jest.fn();
+
+    renderHook(
+      props => {
+        useLayoutSubscription(props.subject$, numSpy);
+        useLayoutSubscription(props.input$, props.subject$);
+      },
+      {
+        initialProps: {
+          input$: num1$,
+          subject$: num2$,
+        },
+      }
+    );
+
+    expect(numSpy).toBeCalledTimes(1);
+  });
 });

packages/observable-hooks/__tests__/use-subscription.spec.ts

@@ -1,7 +1,7 @@
-import { useSubscription } from "../src";
-import { renderHook, act } from "@testing-library/react-hooks";
-import { of, BehaviorSubject, Subject, throwError } from "rxjs";
+import { act, renderHook } from "@testing-library/react-hooks";
 import { useState } from "react";
+import { BehaviorSubject, Subject, of, throwError } from "rxjs";
+import { useSubscription } from "../src";
 
 describe("useSubscription", () => {
   it("should always return the same Subscription after first rendering", () => {
@@ -207,4 +207,25 @@ describe("useSubscription", () => {
     expect(numSpy).toBeCalledTimes(2);
     expect(numSpy).lastCalledWith(2);
   });
+
+  it("should subscribe with Subject", () => {
+    const num1$ = of(1);
+    const num2$ = new Subject();
+    const numSpy = jest.fn();
+
+    renderHook(
+      props => {
+        useSubscription(props.subject$, numSpy);
+        useSubscription(props.input$, props.subject$);
+      },
+      {
+        initialProps: {
+          input$: num1$,
+          subject$: num2$,
+        },
+      }
+    );
+
+    expect(numSpy).toBeCalledTimes(1);
+  });
 });

packages/observable-hooks/src/helpers.ts

@@ -114,4 +114,5 @@ export const useForceUpdate = (): (() => void) => {
  * Prevent React warning when using useLayoutEffect on server.
  */
 export const useIsomorphicLayoutEffect = /* @__PURE__ */ (() =>
+  /* istanbul ignore next */
   typeof window === "undefined" ? useEffect : useLayoutEffect)();

packages/observable-hooks/src/internal/use-subscription-internal.ts

@@ -1,6 +1,12 @@
+import { MutableRefObject, useEffect, useRef } from "react";
 import { Observable, PartialObserver, Subscription } from "rxjs";
 import { useIsomorphicLayoutEffect } from "../helpers";
-import { MutableRefObject, useEffect, useRef } from "react";
+
+interface Observer<T> {
+  next?: (value: T) => void;
+  error?: (err: any) => void;
+  complete?: () => void;
+}
 
 type Args<TInput> = [
   Observable<TInput>, // inputs$
@@ -9,6 +15,15 @@ type Args<TInput> = [
   (() => void) | null | undefined
 ];
 
+const toObserver = <T>(args: Args<T>): Observer<T> =>
+  (args[1] as PartialObserver<T>)?.next
+    ? (args[1] as Observer<T>)
+    : {
+        next: args[1] as Observer<T>["next"],
+        error: args[2] as Observer<T>["error"],
+        complete: args[3] as Observer<T>["complete"],
+      };
+
 /**
  *
  * @template TInput Input value within Observable.
@@ -21,55 +36,45 @@ export function useSubscriptionInternal<TInput>(
   args: Args<TInput>
 ): MutableRefObject<Subscription | undefined> {
   const argsRef = useRef(args);
+  const observerRef = useRef<Observer<TInput>>();
   const subscriptionRef = useRef<Subscription>();
 
   // Update the latest observable and callbacks
   // synchronously after render being committed
   useIsomorphicLayoutEffect(() => {
     argsRef.current = args;
+    observerRef.current = toObserver(args);
   });
 
   useCustomEffect(() => {
     // keep in closure for checking staleness
     const input$ = argsRef.current[0];
 
+    /* istanbul ignore if: Just in case the layoutEffect order is agnostic */
+    if (!observerRef.current) {
+      observerRef.current = toObserver(argsRef.current);
+    }
+
     const subscription = input$.subscribe({
       next: value => {
-        if (input$ !== argsRef.current[0]) {
-          // stale observable
-          return;
-        }
-        const nextObserver =
-          (argsRef.current[1] as PartialObserver<TInput>)?.next ||
-          (argsRef.current[1] as ((value: TInput) => void) | null | undefined);
-        if (nextObserver) {
-          return nextObserver(value);
+        if (input$ === argsRef.current[0]) {
+          observerRef.current!.next?.(value);
         }
+        // else: stale observable
       },
       error: error => {
-        if (input$ !== argsRef.current[0]) {
-          // stale observable
-          return;
+        if (input$ === argsRef.current[0]) {
+          observerRef.current!.error
+            ? observerRef.current!.error(error)
+            : console.error(error);
         }
-        const errorObserver =
-          (argsRef.current[1] as PartialObserver<TInput>)?.error ||
-          argsRef.current[2];
-        if (errorObserver) {
-          return errorObserver(error);
-        }
-        console.error(error);
+        // else: stale observable
       },
       complete: () => {
-        if (input$ !== argsRef.current[0]) {
-          // stale observable
-          return;
-        }
-        const completeObserver =
-          (argsRef.current[1] as PartialObserver<TInput>)?.complete ||
-          argsRef.current[3];
-        if (completeObserver) {
-          return completeObserver();
+        if (input$ === argsRef.current[0]) {
+          observerRef.current!.complete?.();
         }
+        // else: stale observable
       },
     });