feat(utils): startWith -> defaultStart on suspense utils

BREAKING CHANGE: suspend, suspended and switchMapSuspended will only
emit SUSPENSE if the source doesn't emit synchronously

Author: josepot

packages/utils/src/internal-utils.ts

@@ -9,8 +9,8 @@ import {
   mergeMap,
 } from "rxjs/operators"
 
-export const defaultStart = <T>(value: T) => (source$: Observable<T>) =>
-  new Observable<T>((observer) => {
+export const defaultStart = <T, D>(value: D) => (source$: Observable<T>) =>
+  new Observable<T | D>((observer) => {
     let emitted = false
     const subscription = source$.subscribe(
       (x) => {

packages/utils/src/suspend.test.ts

@@ -1,5 +1,6 @@
 import { TestScheduler } from "rxjs/testing"
 import { SUSPENSE } from "@react-rxjs/core"
+import { of } from "rxjs"
 import { suspend } from "./"
 
 const scheduler = () =>
@@ -21,4 +22,18 @@ describe("operators/suspend", () => {
       })
     })
   })
+
+  it("does not prepend the source stream with SUSPENSE when the source is sync", () => {
+    scheduler().run(({ expectObservable }) => {
+      const source = of("a")
+      const expected = "(a|)"
+
+      const suspended = suspend(source)
+
+      expectObservable(suspended).toBe(expected, {
+        s: SUSPENSE,
+        a: "a",
+      })
+    })
+  })
 })

packages/utils/src/suspend.ts

@@ -1,6 +1,6 @@
 import { ObservableInput, from, Observable } from "rxjs"
-import { startWith } from "rxjs/operators"
 import { SUSPENSE } from "@react-rxjs/core"
+import { defaultStart } from "./internal-utils"
 
 /**
  * A RxJS creation operator that prepends a SUSPENSE on the source observable.
@@ -10,4 +10,4 @@ import { SUSPENSE } from "@react-rxjs/core"
 export const suspend: <T>(
   source$: ObservableInput<T>,
 ) => Observable<T | typeof SUSPENSE> = <T>(source$: ObservableInput<T>) =>
-  startWith(SUSPENSE)(from(source$)) as any
+  defaultStart(SUSPENSE)(from(source$)) as any

packages/utils/src/suspended.test.ts

@@ -1,5 +1,6 @@
 import { TestScheduler } from "rxjs/testing"
 import { SUSPENSE } from "@react-rxjs/core"
+import { of } from "rxjs"
 import { suspended } from "./"
 
 const scheduler = () =>
@@ -21,4 +22,18 @@ describe("operators/suspended", () => {
       })
     })
   })
+
+  it("does not prepend the source stream with SUSPENSE when the source is sync", () => {
+    scheduler().run(({ expectObservable }) => {
+      const source = of("a")
+      const expected = "(a|)"
+
+      const result$ = source.pipe(suspended())
+
+      expectObservable(result$).toBe(expected, {
+        s: SUSPENSE,
+        a: "a",
+      })
+    })
+  })
 })

packages/utils/src/switchMapSuspended.test.ts

@@ -1,6 +1,7 @@
 import { TestScheduler } from "rxjs/testing"
 import { SUSPENSE } from "@react-rxjs/core"
 import { switchMapSuspended } from "./"
+import { of } from "rxjs"
 
 const scheduler = () =>
   new TestScheduler((actual, expected) => {
@@ -37,4 +38,19 @@ describe("operators/switchMapSuspended", () => {
       })
     })
   })
+
+  it("does not emits another SUSPENSE when the next inner stream is sync", () => {
+    scheduler().run(({ expectObservable, cold }) => {
+      const source = cold("-x--x")
+      const inner = of("a")
+      const expected = "   -a--a"
+
+      const result$ = source.pipe(switchMapSuspended(() => inner))
+
+      expectObservable(result$).toBe(expected, {
+        s: SUSPENSE,
+        a: "a",
+      })
+    })
+  })
 })