оператор трубопровода ведет себя не так, как ожидается RXJS

Question

Пожалуйста, посмотрите на мой компонент ниже, цель которого состоит в том, чтобы прослушать изменения входа, которые он делает, и затем передать значение родительскому компоненту. Я создал канал, который генерирует только время от времени, и тем самым сводит к минимуму вызовы API, по какой-то причине, хотя я могу видеть через различные операторы console.log, что он идет в канале, он генерирует значение при каждом изменении. Чего мне не хватает:

import {ChangeDetectionStrategy, Component, EventEmitter, Input, OnChanges, OnInit, Output, KeyValueDiffers, DoCheck, KeyValueDiffer} from '@angular/core';
import {BehaviorSubject, Observable, of} from "rxjs";
import {debounceTime, distinctUntilChanged, map, skip, switchMap, takeUntil, tap} from "rxjs/operators";

@Component({
  selector: 'core-ui-typeahead-filter',
  changeDetection: ChangeDetectionStrategy.OnPush,
  templateUrl: './typeahead-filter.component.html',
})
export class TypeaheadFilterComponent implements DoCheck {
  @Input() id: string;
  @Input() name: string;
  @Input() caption: string;
  @Input() placeholder: string;
  @Input() cssClass: string;
  @Input() cssStyle: string;
  @Input() function: any;

  @Input() data: Observable<string[]>;
  differ: any;
  detectChange: string = '';
  // term$ = new BehaviorSubject<string>('');
  text$ = new Observable<string>();

  @Output() onTypeahead: EventEmitter<any> = new EventEmitter<any>();
  @Output() onSelect: EventEmitter<any> = new EventEmitter<any>();


  constructor(private differs: KeyValueDiffers) {
    this.differ = this.differs.find({}).create();
  }

  handleTypeahead = (text$: Observable<string>) =>
    text$.pipe(
      distinctUntilChanged(),
      debounceTime(500),
    ).subscribe((value) => {
      this.onTypeahead.emit(of(value))
    })


handleSelectItem(item) {
  this.onSelect.emit(item);
}

ngDoCheck() {
    const change = this.differ.diff(this);
    if (change) {
      change.forEachChangedItem(item => {
        if (item.key === 'detectChange'){
          console.log('item changed', item)
          this.text$ = of(item.currentValue);
          this.handleTypeahead(this.text$);
        }
      });
    }
  }

}

Дополнительные сведения: на входе присутствует ngModel, связанный с detectChange, когда он изменяется, затем вызывается и выполняется ngDoCheck. Все сделано в наблюдаемых, поэтому в родительском я могу подписаться на входящие события. РЕДАКТИРОВАТЬ ------------------------------------------------- ------------------ Пробовал следующее решение, основанное на моем понимании ответа @ggradnig, к сожалению, он пропускает мою трубу, что-то с ним не так, на самом деле не уверен, что:

 handleTypeahead = (text$: Observable<string>) => {
    this.test.subscribe(this.text$);
    this.test.pipe(
      distinctUntilChanged(),
      debounceTime(500),
      // switchMap(value => text$)
    ).subscribe((value) => {
      tap(console.log('im inside the subscription',value))
      this.onTypeahead.emit(value)
    })
  }



handleSelectItem(item) {
  this.onSelect.emit(item);
}

ngDoCheck() {
    const change = this.differ.diff(this);
    if (change) {
      change.forEachChangedItem(item => {
        if (item.key === 'detectChange'){
          console.log('item changed', item)
          this.text$ = of(item.currentValue);
          this.handleTypeahead(this.test);
        }
      });
    }
  }

}

Answer 1

Вы можете сделать следующее -

export class TypeaheadFilterComponent implements DoCheck {
  @Input() id: string;
  @Input() name: string;
  @Input() caption: string;
  @Input() placeholder: string;
  @Input() cssClass: string;
  @Input() cssStyle: string;
  @Input() function: any;

  @Input() data: Observable<string[]>;
  differ: any;
  detectChange: string = '';
  // term$ = new BehaviorSubject<string>('');
  text$ = new BehaviorSubject<string>('');
  serachTerm$: Observable<string>;

  @Output() onTypeahead: EventEmitter<any> = new EventEmitter<any>();
  @Output() onSelect: EventEmitter<any> = new EventEmitter<any>();


  constructor(private differs: KeyValueDiffers) {
    this.differ = this.differs.find({}).create();
  }

  // handleTypeahead = (text$: Observable<string>) =>
  //   text$.pipe(
  //     distinctUntilChanged(),
  //     debounceTime(500),
  //   ).subscribe((value) => {
  //     this.onTypeahead.emit(of(value))
  //   })

  ngOnInit() {

    this.serachTerm$ = this.text$
                           .pipe(
                            distinctUntilChanged(),
                            debounceTime(500),
                            //filter(), //use filter operator if your logic wants to ignore certain string like empty/null
                            tap(s => this.onTypeahead.emit(s))
                           );
  } 

handleSelectItem(item) {
  this.onSelect.emit(item);
}

ngDoCheck() {
    const change = this.differ.diff(this);
    if (change) {
      change.forEachChangedItem(item => {
        if (item.key === 'detectChange'){
          console.log('item changed', item)
          this.text$.next(item.currentValue);          
        }
      });
    }
  }

}

Теперь в нижней части вашего шаблона поместите следующую строку -

<ng-container *ngIf="searchTerm$ | async"></ng-container>

Наличие этой строки сохранит ваш код компонента в свободной формеуправление подпиской [т.е. не нужно подписываться / отписываться];об этом позаботится асинхронная труба.