Angular Structural Directive for Handling Observables Better than NgIf & AsyncPipe

observe.directive.spec.ts

import {Component, DebugElement} from '@angular/core';
import {ComponentFixture, TestBed} from '@angular/core/testing';
import {By} from '@angular/platform-browser';
import {ObserveDirective} from './observe.directive';
import {BehaviorSubject, Observable, of, throwError} from 'rxjs';

@Component({
  selector: 'test-component',
  template: `
    <span id="next-value" *observe="value$ as value; before loading; error error;">
      {{ value }}
    </span>
    <ng-template #loading>
      <span id="loading-text">Loading</span>
    </ng-template>
    <ng-template #error let-error>
      <span id="error-value">{{ error }}</span>
    </ng-template>
  `
})
class TestComponent {
  value$: Observable<string>;
}

describe('ObserveDirective', () => {

  let fixture: ComponentFixture<TestComponent>;
  let component: TestComponent;
  let el: DebugElement;

  beforeEach(() => {
    TestBed.configureTestingModule({
      declarations: [
        TestComponent,
        ObserveDirective
      ]
    });
    fixture = TestBed.createComponent(TestComponent);
    component = fixture.componentInstance;
    el = fixture.debugElement;
  });

  it('should create host', () => {
    expect(component).toBeDefined();
  });

  describe('when no observable provided', () => {
    it('should only render before-template', () => {
      fixture.detectChanges();
      expect(el.query(By.css('#loading-text')).nativeElement.textContent).toEqual('Loading');
      expect(el.query(By.css('#next-value'))).toBeNull();
      expect(el.query(By.css('#error-value'))).toBeNull();
    });
  });

  describe('when observable provided', () => {
    it('should render values into next-template', () => {
      component.value$ = of('First');
      fixture.detectChanges();
      expect(el.query(By.css('#loading-text'))).toBeNull();
      expect(el.query(By.css('#error-value'))).toBeNull();
      expect(el.query(By.css('#next-value')).nativeElement.textContent.trim()).toEqual('First');
      const sink = new BehaviorSubject('Second');
      component.value$ = sink;
      fixture.detectChanges();
      expect(el.query(By.css('#next-value')).nativeElement.textContent.trim()).toEqual('Second');
      sink.next('Third');
      fixture.detectChanges();
      expect(el.query(By.css('#next-value')).nativeElement.textContent.trim()).toEqual('Third');
    });

    it('should render errors into error-template', () => {
      component.value$ = throwError('TestError');
      fixture.detectChanges();
      expect(el.query(By.css('#loading-text'))).toBeNull();
      expect(el.query(By.css('#next-value'))).toBeNull();
      expect(el.query(By.css('#error-value')).nativeElement.textContent.trim()).toEqual('TestError');
    });

    it('should replace next-template with error-template upon error', () => {
      const sink = new BehaviorSubject('Success');
      component.value$ = sink;
      fixture.detectChanges();
      expect(el.query(By.css('#loading-text'))).toBeNull();
      expect(el.query(By.css('#error-value'))).toBeNull();
      expect(el.query(By.css('#next-value')).nativeElement.textContent.trim()).toEqual('Success');
      sink.error('Error');
      fixture.detectChanges();
      expect(el.query(By.css('#loading-text'))).toBeNull();
      expect(el.query(By.css('#next-value'))).toBeNull();
      expect(el.query(By.css('#error-value')).nativeElement.textContent.trim()).toEqual('Error');
    });

    it('should wait for OnInit to subscribe', () => {
      const sink = new BehaviorSubject('Success');
      component.value$ = sink;
      expect(sink.observers.length).toBe(0);
      fixture.detectChanges(); // first detection fires OnInit
      expect(sink.observers.length).toBe(1);
    });

    it('should unsubscribe upon destruction', () => {
      const sink = new BehaviorSubject('Success');
      component.value$ = sink;
      fixture.detectChanges();
      expect(sink.observers.length).toBe(1);
      fixture.destroy();
      expect(sink.observers.length).toBe(0);
    });
    
    it('should not resubscribe to same source', () => {
      let subscribeCount = 0;
      const observable = defer(() => {
        subscribeCount++;
        return of('Success');
      });
      component.value$ = observable;
      fixture.detectChanges();
      component.value$ = observable;
      fixture.detectChanges();
      expect(subscribeCount).toBe(1);
    });

    it('should unsubscribe upon changed source', () => {
      const sink = new BehaviorSubject('Success');
      component.value$ = sink;
      fixture.detectChanges();
      expect(sink.observers.length).toBe(1);
      const otherSink = new BehaviorSubject('Another Success');
      component.value$ = otherSink;
      fixture.detectChanges();
      expect(sink.observers.length).toBe(0);
      expect(otherSink.observers.length).toBe(1);
      component.value$ = undefined;
      fixture.detectChanges();
      expect(otherSink.observers.length).toBe(0);
    });

  });
});

observe.directive.ts

import {
  Directive, Input, TemplateRef, ViewContainerRef,
  OnDestroy, OnInit, ChangeDetectorRef
} from '@angular/core'
import { Observable, Subject, AsyncSubject } from "rxjs";
import { takeUntil, concatMapTo, finalize } from "rxjs/operators";

export interface ObserveContext<T> {
  $implicit: T;
  observe: T;
}

export interface ErrorContext {
  $implicit: Error;
}

@Directive({
  selector: "[observe]"
})
export class ObserveDirective<T> implements OnDestroy, OnInit {
  private errorRef: TemplateRef<ErrorContext>
  private beforeRef: TemplateRef<null>
  private unsubscribe = new Subject<boolean>()
  private init = new AsyncSubject<void>()
  private source: Observable<T>

  constructor(
    private view: ViewContainerRef,
    private nextRef: TemplateRef<ObserveContext<T>>,
    private changes: ChangeDetectorRef
  ) {}

  @Input()
  set observe(source: Observable<T>) {
    if (!source) {
      this.source = source;
      this.unsubscribe.next(true);
      return;
    }
    if (this.source && this.source !== source) {
      this.unsubscribe.next(true);
    }
    this.source = source;
    this.showBefore()
    this.init.pipe(
      concatMapTo(source),
      takeUntil(this.unsubscribe)
    ).subscribe(value => {
      this.view.clear()
      this.view.createEmbeddedView(this.nextRef, {$implicit: value, observe: value})
      this.changes.markForCheck()
    }, error => {
      if (this.errorRef) {
       this.view.clear()
       this.view.createEmbeddedView(this.errorRef, {$implicit: error})
       this.changes.markForCheck()
      }
    })
  }

  @Input()
  set observeError(ref: TemplateRef<ErrorContext>) {
    this.errorRef = ref
  }

  @Input()
  set observeBefore(ref: TemplateRef<null>) {
    this.beforeRef = ref
  }

  ngOnDestroy() {
    this.unsubscribe.next(true)
  }

  ngOnInit() {
    this.showBefore()
    this.init.next()
    this.init.complete()
  }

  private showBefore(): void {
    if (this.beforeRef) {
      this.view.clear()
      this.view.createEmbeddedView(this.beforeRef)
    }
  }
}

usage.html

<p *observe="users$ as users; before loadingTemplate; error errorTemplate">
  There are {{ users.length }} online.
</p>
<ng-template #loadingTemplate>
  <p>Loading ...</p>
</ng-template>
<ng-template #errorTemplate let-error>
  <p>{{ error }}</p>
</ng-template>