Ограничение сетевых запросов на наблюдаемые через оператор RxJS

Question

У меня есть данные, поступающие в компонент моего приложения Angular через сетевой запрос.Это observable и выглядит так:

    this.streamFiltersService.getByFilters(
        this.page - 1, this.pagesize, this.currentStage, this.language = filters['language'], this.location = filters['location'],
        this.zipcode = filters['zip'], this.firstName = filters['firstName'], this.lastName = filters['lastName'],
        this.branch = filters['branch'], fn);

Здесь есть обратный вызов fn, который выглядит так:

    let fn = resRecordsData => {
        this.records = resRecordsData;
    };

Проблема, с которой я работаюДело в том, что по мере того, как многочисленные компоненты фильтра входят в компонент, я получаю несколько сетевых запросов - и для пользователя данные экрана меняются несколько раз.Кроме того, поскольку это асинхронно, иногда один из первых запросов возвращается последним, и фильтры не применяются на экране, который видит пользователь.

Полная функция выглядит следующим образом:

public onFilterReceived(values)
{
    let filters = {};

    if (values) {
        filters = values;
    }

    this.route.params.subscribe(
        (params: any) => {
            this.page = params['page'];
        }
    );

    console.log('onFilterReceived() firing...');

    let fn = resRecordsData => {
        this.records = resRecordsData;
    };

    this.streamFiltersService.getByFilters(
        this.page - 1, this.pagesize, this.currentStage, this.language = filters['language'], this.location = filters['location'],
        this.zipcode = filters['zip'], this.firstName = filters['firstName'], this.lastName = filters['lastName'],
        this.branch = filters['branch'], fn);
}

Как продолжение, когда я добавляю console.log к обратному вызову в onFilterReceived(), вот так:

let fn = async resRecordsData => {
    console.log('records: ', resRecordsData);
    this.records = await resRecordsData;
};

Что выводит на консоль так:

records:  {ok: true, status: 200, statusText: "OK", data: Array(1), count: 1}
records:  {ok: true, status: 200, statusText: "OK", data: Array(12), count: 115}
records:  {ok: true, status: 200, statusText: "OK", data: Array(1), count: 1}
records:  {ok: true, status: 200, statusText: "OK", data: Array(1), count: 1}
records:  {ok: true, status: 200, statusText: "OK", data: Array(1), count: 1}
records:  {ok: true, status: 200, statusText: "OK", data: Array(12), count: 115}
records:  {ok: true, status: 200, statusText: "OK", data: Array(12), count: 115}
records:  {ok: true, status: 200, statusText: "OK", data: Array(12), count: 115}
records:  {ok: true, status: 200, statusText: "OK", data: Array(12), count: 115}
records:  {ok: true, status: 200, statusText: "OK", data: Array(12), count: 115}

Обратите внимание, что, поскольку я применил фильтр, правильным значением является Массив (1).Как видите, поскольку эти вызовы асинхронны, они возвращаются не в порядке.В идеале мне нужен один вызов, который приводит к следующему:

records:  {ok: true, status: 200, statusText: "OK", data: Array(1), count: 1}

Основываясь на предложении ниже, я попытался объединить предложенные операторы в обратный вызов onFilterReceived(), например:

public async onFilterReceived(values)
{
    let filters = {};

    if (values) {
        filters = values;
    }

    this.route.params.subscribe(
        (params: any) => {
            this.page = params['page'];
        }
    );

    console.log('onFilterReceived() firing...');

    let fn = async resRecordsData => {
        await resRecordsData
        .distinctUntilChanged()
        .debounceTime(1000)
        .switchMap( resRecordsData => {
            this.records = resRecordsData;
        });
        console.log('records: ', this.records);
    };

    this.streamFiltersService.getByFilters(
        this.page - 1, this.pagesize, this.currentStage, this.language = filters['language'], this.location = filters['location'],
        this.zipcode = filters['zip'], this.firstName = filters['firstName'], this.lastName = filters['lastName'],
        this.branch = filters['branch'], fn);
}

... но я получаю сообщение об ошибке:

error_handler.js: 60 Ошибка: Uncaught (в обещании): TypeError: resRecordsData.distinctUntilChanged не является функцией

Answer 1

Так что, если я правильно понимаю ваши проблемы:

• вы поддерживаете глобальное состояние (this.records) с помощью асинхронных обратных вызовов
• http-запросы запускаются, но возвращаются неупорядоченными.

Это очень похоже на часто повторяющийся пример автозаполнения с rxjs .

Код, который вам понадобится, будет примерно таким:

$filterChanges
  .distinctUntilchanged()
  .debounceTime(1000)
  .switchMap(filters => getByFilters(.....))
  .subscribe(records => {
    // modify global state with the records
  });

Итак, вы сначала берете поток своих фильтров и их изменений - например, { page, pagesize, currentStage, language }.Возможно, вам придется добавить compareFunc к distinctUntilChanged для объекта filters, поскольку он не является примитивным.

Ниже приведен грубый пример неангулярной реализации, который поможет вам:

/* make our html filter elements observable */
var carTypeFilter = document.getElementById('cars');
var hornsCheckbox = document.getElementById('horns');

var $carsFilter = Rx.Observable.fromEvent(carTypeFilter, 'change')
  .map(evt => ({ filter: 'car', value: evt.target.value }));
var $hornsFilter = Rx.Observable.fromEvent(hornsCheckbox, 'change')
  .map(evt => ({ filter: 'horns', value: evt.target.checked }));

/* we want to have one value containing the current logical grouping of filters */
var $filtersCombined = Rx.Observable.merge($carsFilter,$hornsFilter)
  .scan((allCurrentFilters, currFilterItem) => {
    allCurrentFilters[currFilterItem.filter] = currFilterItem.value;
    return allCurrentFilters;
  }, {});


var $debouncedFilterChanges = $filtersCombined
  /* for distinctUntilChanged to work
  you would need to have a compareFunction to evaluate
  if all the key:value pairs of the filters are the same
  not going to do that in this example; it will only filter out 
  filter sets after franctic clicking but only result in an additional
  request being done to the backend */
  //  .distinctUntilChanged()
  /* for every distinct filterSetChange debounce it so 
  we know that the user has stopped fiddling with the inputs */
  .debounceTime(500);

var $filteredServerResults = $debouncedFilterChanges
  .switchMap(filters => getFilteredData(filters.car, filters.horns));

$filteredServerResults.subscribe(data => {
  document.getElementById('results').innerText = JSON.stringify(data);
});


/*
mock function which simulates doing async call to server with slow respons
*/
function getFilteredData(car, horns){
  //note that the car or horns can be undefined
  return Rx.Observable.of(`retrieved values for: ${car}-${horns}`)
    .delay(1500); // mimic slow response, user can have already asked newer results
}

<!DOCTYPE html>


  
  
  JS Bin
  



  
    
      Select one...
      Volvo
      Saab
      Mercedes
      Audi
    
    Car
  
  
    
    HasHorns
  
  results...