RxAsync Directive: Elegant Async Loading, Reload, and Retry Handling in Angular

This article introduces the rxAsync directive, which leverages ObservableInput to elegantly manage asynchronous operations in Angular components, handling loading, reload, and retry states without complex imperative code.

It outlines four key considerations when solving this problem: (1) three request initiation scenarios (initial load, user‑triggered reload, automatic retry on error); (2) dynamic rendering based on loading, success, and error states; (3) reacting to parameter changes while ignoring retry‑count changes; and (4) supporting both internal and external reload triggers.

The implementation is shown below, using a TypeScript class decorated with @Directive . The class defines observable inputs for context, fetcher, parameters, refetch, and retry times, and combines them with combineLatest , withLatestFrom , and other RxJS operators to drive the async workflow, automatically updating the view and handling errors.

@Directive({
  selector: '[rxAsync]',
})
export class AsyncDirective
implements OnInit, OnDestroy {
  @ObservableInput()
  @Input('rxAsyncContext')
  private context$!: Observable
;

  @ObservableInput()
  @Input('rxAsyncFetcher')
  private fetcher$!: Observable
>>;

  @ObservableInput()
  @Input('rxAsyncParams')
  private params$!: Observable
;

  @Input('rxAsyncRefetch')
  private refetch$$ = new Subject
();

  @ObservableInput()
  @Input('rxAsyncRetryTimes')
  private retryTimes$!: Observable
;

  private destroy$$ = new Subject
();
  private reload$$ = new Subject
();
  private context = { reload: this.reload.bind(this) } as IAsyncDirectiveContext
;
  private viewRef: Nullable
;
  private sub: Nullable
;

  constructor(private templateRef: TemplateRef
, private viewContainerRef: ViewContainerRef) {}

  reload() { this.reload$$.next(); }

  ngOnInit() {
    combineLatest([
      this.context$, this.fetcher$, this.params$,
      this.refetch$$.pipe(startWith(null)),
      this.reload$$.pipe(startWith(null))
    ])
    .pipe(
      takeUntil(this.destroy$$),
      withLatestFrom(this.retryTimes$)
    )
    .subscribe(([[context, fetcher, params], retryTimes]) => {
      this.disposeSub();
      Object.assign(this.context, { loading: true, error: null });
      this.sub = fetcher.call(context, params).pipe(
        retry(retryTimes),
        finalize(() => {
          this.context.loading = false;
          if (this.viewRef) this.viewRef.detectChanges();
        })
      ).subscribe(
        data => this.context.$implicit = data,
        error => this.context.error = error
      );
      if (this.viewRef) return this.viewRef.markForCheck();
      this.viewRef = this.viewContainerRef.createEmbeddedView(this.templateRef, this.context);
    });
  }

  ngOnDestroy() {
    this.disposeSub();
    this.destroy$$.next();
    this.destroy$$.complete();
    if (this.viewRef) { this.viewRef.destroy(); this.viewRef = null; }
  }

  private disposeSub() { if (this.sub) { this.sub.unsubscribe(); this.sub = null; } }
}

A usage example demonstrates a component applying the *rxAsync structural directive, binding a fetchTodo function, exposing loading , error , reload , and the fetched todo data, and allowing an external refetch via a Subject .

@Component({
  selector: 'rx-async-directive-demo',
  template: `
Refetch (Outside rxAsync)
Reload
loading: {{ loading }} error: {{ error | json }}
todo: {{ todo | json }}
`,
  preserveWhitespaces: false,
  changeDetection: ChangeDetectionStrategy.OnPush,
})
class AsyncDirectiveComponent {
  @Input() todoId = 1;
  @Input() retryTimes = 0;
  refetch$$ = new Subject
();
  constructor(private http: HttpClient) {}
  fetchTodo(todoId: string) {
    return typeof todoId === 'number'
      ? this.http.get('//jsonplaceholder.typicode.com/todos/' + todoId)
      : EMPTY;
  }
}

Overall, the rxAsync directive provides a concise, reactive solution for common asynchronous UI patterns in frontend development, reducing boilerplate and improving readability.