create-effect.ts

import { assertInInjectionContext, DestroyRef, inject, Injector, isDevMode, isSignal, type Signal } from '@angular/core';
import { takeUntilDestroyed, toObservable } from '@angular/core/rxjs-interop';
import { isObservable, Observable, of, retry, type RetryConfig, Subject, Subscription } from 'rxjs';

export type CreateEffectOptions = {
  injector?: Injector,
  /**
   * @param retryOnError
   * Set to 'false' to disable retrying on error.
   * Otherwise, generated effect will use `retry()`.
   * You can pass `RetryConfig` object here to configure `retry()` operator.
   */
  retryOnError?: boolean | RetryConfig,
};

/**
 * This code is copied from NgRx ComponentStore and edited to add `takeUntilDestroyed()` and to resubscribe on errors.
 * Credits: NgRx Team
 * https://ngrx.io/
 * Source: https://github.com/ngrx/platform/blob/main/modules/component-store/src/component-store.ts#L382
 * Docs:
 * https://ngrx.io/guide/component-store/effect#effect-method
 */
export function createEffect<
  ProvidedType = void,
  OriginType extends | Observable<ProvidedType> | unknown = Observable<ProvidedType>,
  ObservableType = OriginType extends Observable<infer A> ? A : never,
  ReturnType = ProvidedType | ObservableType extends void
    ? (
      observableOrValue?: ObservableType | Observable<ObservableType> | Signal<ObservableType>
    ) => Subscription
    : (
      observableOrValue: ObservableType | Observable<ObservableType> | Signal<ObservableType>
    ) => Subscription
>(generator: (origin$: OriginType) => Observable<unknown>, options?: CreateEffectOptions): ReturnType {

  if (!options?.injector && isDevMode()) {
    assertInInjectionContext(createEffect);
  }

  const injector = options?.injector ?? inject(Injector);
  const destroyRef = injector.get(DestroyRef);
  const origin$ = new Subject<ObservableType>();
  const retryOnError = options?.retryOnError ?? true;
  const retryConfig = (typeof options?.retryOnError === 'object' && options?.retryOnError) ? options?.retryOnError : {} as RetryConfig;

  if (retryOnError) {
    generator(origin$ as OriginType).pipe(
      retry(retryConfig),
      takeUntilDestroyed(destroyRef)
    ).subscribe();
  } else {
    generator(origin$ as OriginType).pipe(
      takeUntilDestroyed(destroyRef)
    ).subscribe();
  }

  return ((
    observableOrValue?: ObservableType | Observable<ObservableType> | Signal<ObservableType>
  ): Subscription => {
    const observable$ = isObservable(observableOrValue)
      ? observableOrValue
      : (isSignal(observableOrValue)
          ? toObservable(observableOrValue, { injector })
          : of(observableOrValue)
      );
    if (retryOnError) {
      return observable$.pipe(
        retry(retryConfig),
        takeUntilDestroyed(destroyRef)
      ).subscribe((value) => {
        origin$.next(value as ObservableType);
      });
    } else {
      return observable$.pipe(
        takeUntilDestroyed(destroyRef)
      ).subscribe((value) => {
        origin$.next(value as ObservableType);
      });
    }
  }) as unknown as ReturnType;
}

createEffect is a standalone version of NgRx ComponentStore Effect

---

From ComponentStore documentation:

• Effects isolate side effects from components, allowing for more pure components that select state and trigger updates and/or effects in ComponentStore(s).
• Effects are Observables listening for the inputs and piping them through the "prescription".
• Those inputs can either be values or Observables of values.
• Effects perform tasks, which are synchronous or asynchronous.

---

Usage

@Component({})
export class Some {
  log = createEffect<number>(
    pipe(
      map((value) => value * 2),
      tap((v) => console.log('double:', v)),
    ),
  );

  ngOnInit() {
    // start the effect
    this.log(interval(1000));
  }
}

Injection Context

createEffect accepts an optional Injector so we can call createEffect outside of an Injection Context.

@Component({})
export class Some {
  // 1. setup an Input; we know that Input isn't resolved in constructor
  @Input() multiplier = 2;

  // 2. grab the Injector
  private injector = inject(Injector);

  ngOnInit() {
    // 3. create log effect in ngOnInit; where Input is resolved
    const log = createEffect<number>(
      pipe(
        map((value) => value * this.multiplier),
        tap((v) => console.log('multiply: ', v))
      ),
      // 4. pass in the injector
      { injector: this.injector },
    );

    // 5. start the effect
    log(interval(1000));
  }
}

Resubscribe on errors

By default, createEffect() will re-subscribe on errors, using retry() operator.
This behavior can be configured or turned off, using optional second argument:

@Component({})
export class Example {
  // Will not resubscribe on error
  private loadProducts = createEffect<string>(
    (_) => _.pipe(switchMap((id) => this.api.loadProducts(id))),
    { retryOnEror: false },
  );

  // Will resubscribe on error with a delay, not more than 3 times
  private loadProducts = createEffect<string>(
    (_) => _.pipe(switchMap((id) => this.api.loadProducts(id))),
    { retryOnEror: { count: 3, delay: 500 } },
  );
}