RxJs: шаблон для наблюдаемых результатов поиска

Question

Мой сценарий - это классическая веб-страница с формой поиска и списком результатов. Я хочу инкапсулировать поведение загрузки результатов в Observable.

Вот что я сейчас делаю в TypeScript:

function loadResults(query): Observable<T[]> {}

const results = new Subject<ResultEvent<T[]>>();

const results: Observable<ResultEvent<T[]>> =
  form.valueChanges
    .distinctUntilChanged()
    .do(() => results.next(ResultEvent.pending()))
    .switchMap(query => loadResults(query))
    .subscribe({
      next: (data: T[]) => results.next(ResultEvent.present(data)),
      error: err => results.next(ResultEvent.failed(err)),
    });

Идея в том, что results всегда содержит текущее состояние поиска: либо pending, present или же failed, Когда запрос изменяется, результат устанавливается на pendingи когда служба возвращает данные, результат устанавливается на present,

Что мне не нравится в этом решении, так это явный вызов subscribe(), Я бы предпочел простой Observable которые могут быть подписаны как отписавшиеся от (например, в Angular с async труба), без создания явной подписки. Побочные эффекты в do тоже кажутся довольно хакерскими.

const results: Obserbable<ResultEvent<T[]>> = 
  form.valueChanges.distinctUntilChanged()
  . /* here be dragons */;

Спасибо за любые советы и идеи!

Answer 1

Я думаю, что вы хотите что-то вроде этого:

const results$ = form.valueChanges
  // This is up to you, but with user input it might make sense to
  // give it just a little bit of time before we hit the server since
  // most user input will be more than a single character.
  //.debounceTime(100)

  .distinctUntilChanged()

  // Using switchMap guarantees that the inner observable will be
  // cancelled if the input changed while we are still waiting for
  // a result. Newer is always better!
  .switchMap(query => loadResults(query)
    // If we get data, we use it.
    .map(results => ResultEvent.present(results))

    // We catch errors and turn them into a failure event.
    .catch(err => Observable.of(ResultEvent.failed(err)))

    // Whatever happens, first things first.
    .startWith(ResultEvent.pending())
  );

Я бы также подумал о добавлении debounceTime там, кстати.

Вот фрагмент, который вы можете скопировать и вставить на https://rxviz.com/ чтобы увидеть его в действии (к сожалению, функция ссылок на него больше не работает). Обязательно установите временное окно примерно на 10 секунд.

const ResultEvent = {
  pending: () => 'Pending',
  failed: err => 'Error: ' + err,
  present: data => 'Data: ' + data,
};

const loadResults = query => query === 2
  ? Rx.Observable.of(null).delay(500).switchMap(() => Rx.Observable.throw('Oops'))
  : Rx.Observable.of(42).delay(500)

const input$ = Rx.Observable.timer(0, 2000).take(4);

input$.switchMap(query => loadResults(query)
  .map(data => ResultEvent.present(data))
  .catch(err => Rx.Observable.of(ResultEvent.failed(err)))
  .startWith(ResultEvent.pending())
)