feat(core): implementation of rxResource() for interop

packages/core/rxjs-interop/src/rx_resource.ts

@@ -0,0 +1,69 @@
+/**
+ * @license
+ * Copyright Google LLC All Rights Reserved.
+ *
+ * Use of this source code is governed by an MIT-style license that can be
+ * found in the LICENSE file at https://angular.dev/license
+ */
+
+import {ResourceStatus, ValueEqualityFn, WritableResource} from '@angular/core';
+import {BaseWritableResource} from '@angular/core/src/resource/resource';
+import {catchError, merge, Observable, of, Subject, switchMap, take} from 'rxjs';
+import {takeUntilDestroyed} from './take_until_destroyed';
+
+export interface RxResourceOptions<R, T> {
+  source: Observable<R>;
+  loader: (req: Exclude<NoInfer<R>, undefined>) => Observable<T>;
+  equal?: ValueEqualityFn<T>;
+}
+
+export function rxResource<R, T>({
+  source,
+  loader,
+  equal,
+}: RxResourceOptions<R, T>): WritableResource<T> {
+  return new RxResourceImpl(source, loader, equal);
+}
+
+export function toResource<T>(observable: Observable<T>): WritableResource<T> {
+  return rxResource({
+    source: of(observable),
+    loader: (r) => r,
+  });
+}
+
+const LOCAL = Symbol('LOCAL') as any;
+class RxResourceImpl<R, T> extends BaseWritableResource<T> {
+  private local$ = new Subject<R>();
+
+  constructor(
+    source: Observable<R>,
+    loadFn: (request: Exclude<R, undefined>) => Observable<T>,
+    equal: ValueEqualityFn<T> | undefined,
+  ) {
+    super(equal);
+    merge(source, this.local$).pipe(
+      switchMap((req) => {
+        if (req === LOCAL) {
+          return of();
+        } else if (req === undefined) {
+          this.setValueState(ResourceStatus.Idle);
+          return of();
+        }
+
+        return loadFn(req as Exclude<R, undefined>).pipe(
+          take(1),
+          catchError((err) => {
+            this.setErrorState(err);
+            return of();
+          }),
+        );
+      }),
+      takeUntilDestroyed(),
+    );
+  }
+
+  protected override maybeCancelLoad(): void {
+    this.local$.next(LOCAL);
+  }
+}