Function rxResource

packages/core/rxjs-interop/src/rx_resource.ts:62–145

(opts: RxResourceOptions<T, R>)

 */
export function rxResource<T, R>(opts: RxResourceOptions<T, R>): ResourceRef<T | undefined>;
export function rxResource<T, R>(opts: RxResourceOptions<T, R>): ResourceRef<T | undefined> {
  if (ngDevMode && !opts?.injector) {
    assertInInjectionContext(rxResource);
  }
  return resource<T, R>({
    ...opts,
    loader: undefined,
    stream: (params) => {
      let sub: Subscription | undefined;

      // `abort` can fire synchronously while the subscription is not initialized yet.
      // Use this flag to unsubscribe immediately once `sub` exists.
      let aborted = false;

      // Start off stream as undefined.
      const stream = signal<ResourceStreamItem<T>>({value: undefined as T});
      const {resolve, promise} = promiseWithResolvers<Signal<ResourceStreamItem<T>>>();
      let hasResolved = false;

      function resolveOnce(): void {
        if (!hasResolved) {
          hasResolved = true;
          resolve(stream);
        }
      }

      // Track the abort listener so it can be removed if the Observable completes (as a memory
      // optimization).
      const onAbort = () => {
        aborted = true;
        sub?.unsubscribe();
        // Remove the listener immediately since unsubscribe won't trigger the subscription's
        // error/complete handlers. This ensures the promise resolves and PendingTask is released.
        params.abortSignal.removeEventListener('abort', onAbort);
        // Resolve the promise with the current stream state if it hasn't been resolved yet.
        // This ensures the PendingTask created for this request is released.
        resolveOnce();
      };
      params.abortSignal.addEventListener('abort', onAbort);

      function send(value: ResourceStreamItem<T>): void {
        stream.set(value);
        resolveOnce();
      }

      const streamFn = opts.stream;
      if (streamFn === undefined) {
        throw new ɵRuntimeError(
          ɵRuntimeErrorCode.MUST_PROVIDE_STREAM_OPTION,
          ngDevMode && `Must provide \`stream\` option.`,
        );
      }

      sub = streamFn(params).subscribe({
        next: (value) => send({value}),
        error: (error: unknown) => {
          send({error: encapsulateResourceError(error)});
          params.abortSignal.removeEventListener('abort', onAbort);