Move 'behavior' to be a method on ObservableScope

src/state/ObservableScope.ts

@@ -5,10 +5,20 @@ SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial
 Please see LICENSE in the repository root for full details.
 */
 
-import { type Observable, Subject, takeUntil } from "rxjs";
+import {
+  BehaviorSubject,
+  distinctUntilChanged,
+  type Observable,
+  Subject,
+  takeUntil,
+} from "rxjs";
+
+import { type Behavior } from "./Behavior";
 
 type MonoTypeOperator = <T>(o: Observable<T>) => Observable<T>;
 
+const nothing = Symbol("nothing");
+
 /**
  * A scope which limits the execution lifetime of its bound Observables.
  */
@@ -25,6 +35,33 @@ export class ObservableScope {
     return this.bindImpl;
   }
 
+  /**
+   * Converts an Observable to a Behavior. If no initial value is specified, the
+   * Observable must synchronously emit an initial value.
+   */
+  public behavior<T>(
+    setValue$: Observable<T>,
+    initialValue: T | typeof nothing = nothing,
+  ): Behavior<T> {
+    const subject$ = new BehaviorSubject(initialValue);
+    // Push values from the Observable into the BehaviorSubject.
+    // BehaviorSubjects have an undesirable feature where if you call 'complete',
+    // they will no longer re-emit their current value upon subscription. We want
+    // to support Observables that complete (for example `of({})`), so we have to
+    // take care to not propagate the completion event.
+    setValue$.pipe(this.bind(), distinctUntilChanged()).subscribe({
+      next(value) {
+        subject$.next(value);
+      },
+      error(err: unknown) {
+        subject$.error(err);
+      },
+    });
+    if (subject$.value === nothing)
+      throw new Error("Behavior failed to synchronously emit an initial value");
+    return subject$ as Behavior<T>;
+  }
+
   /**
    * Ends the scope, causing any bound Observables to complete.
    */