LironHazan

  notNullish<T>(input: null | undefined | T): boolean {
    return input !== null && input !== undefined;
  }

  registerToggleCall() {
    Interactor.onTogglingSomthing()
      .pipe(distinctUntilChanged(), filter(this.notNullish))
      .subscribe((_state: boolean) => {
        this.state = _state;
      });

LironHazan

  private searchQuery$ = new BehaviorSubject('');
  result$: Observable<string>; 

  ngOnInit(): void {
    this.result$ = this.searchQuery$.pipe(
      debounceTime(300), // will drop some values if they occur too frequently  
      switchMap(() =>this.service.fetchSomething(
        this.searchQuery$.getValue()) // will switch to the new inner observable when source emits and cancel the previus one
      )
    );

LironHazan

private myStream = new Subject();
private myStream$ = this.myStream.asObservable();

ngOnInit() {

  // Will only fire after 100 ms.
  this.myStream$
    .pipe(auditTime(100))
    .subscribe(e => console.log('will only fire after 100 miliseconds');
  

LironHazan

  public getHttpDistinctResponse(httpRequest$: Observable<any>): Observable<any> {
    let cachedResponseStr = null;
    return httpRequest$.pipe(
      filter((response) => {
        const currentResponseStr = JSON.stringify(response);
        const areEquals = cachedResponseStr === currentResponseStr;
        if (!areEquals) {
          cachedResponseStr = currentResponseStr;
        }
        return !areEquals;

LironHazan

interactionSubject = new Subject<void>();
ngOnInit() {
  this.interactionSubject.asObservable().pipe(
    switchMap((data) => {
      this.server.fetchData(data).pipe(
        catchError((err) => {
          // handle error
        })
      )
    })

LironHazan

export function pollingOnResolved(httpRequest$: Observable<any>, delayMs = 0): Observable<any> {
  const polling$ = new BehaviorSubject({});
  const rePolling$ = 
    of('').pipe(
    delay(delayMs),
    tap(() => polling$.next({})),
    skip(1)
  );
  const httpPolling$ = concat(httpRequest$, rePolling$);
  return polling$.pipe(switchMap(() => httpPolling$));

LironHazan

  private async getTheBand(name: string): Promise<Artist> {
    const artist = await import('../../../wasm/pkg/rust_wasm_part');
    return artist.query_band(name);
  }

  getTheBand$(name): Observable<Artist> {
    return defer(() => this.getTheBand(name));
  }

LironHazan

  registerTransitionEnd<T extends {propertyName}>(el: FromEventTarget<T>, propName: string) {
    // Returns an Observable from DOM event 
    fromEvent(el, 'transitionend')
      .pipe(
        filter(({ propertyName }: T) => propertyName === propName) // Will only emit value for the wanted css prop name e.g. height 
      )
      .subscribe((_) => {
        this.transitionEnd.emit();
      });
  }

LironHazan

  private unsubscribeDataFetch: Subject<void> = new Subject<void>();

   // polling data till we get what we need and stop 
   fetchData() {
    this.dataService
      .pipe(takeUntil(this.unsubscribeDataFetch))
      .subscribe((data) => actOnData(data));
  }
    
 actOnData(data) {

LironHazan

import {BehaviorSubject, Observable} from 'rxjs';
export class StoreService<T> {
  private readonly state$: BehaviorSubject<T>;
  protected get state(): T {
    return this.state$.getValue();
  }
  constructor(initialState: T) {
    this.state$ = new BehaviorSubject<T>(initialState);
  }
  protected getState(): Observable<T> {