Using SwitchMap() to Handle Canceling Previous Requests

Question

I have a feature in my Angular2 app where a user can select various filters to filter data being returned in a grid view. This getting and filtering of data is working. However, I need to find a way - preferably using RxJS and something like switchMap() to cancel requests when changes are made to the filter selections -- so that only the most recent request goes over the wire.

I've been having difficulty getting this to work. So I first need to determine if my current configuration actually is an RxJS observable, and then, where to plug in an operator like switchMap().

This is what my code looks like:

private sendRequest = _.debounce((languageFilter, locationFilter, zipFilter, firstNameFilter, lastNameFilter, branchFilter) =>
{
    this.filtersService.getByFilters(
        this.page - 1, this.pagesize, this.currentStage, this.language = languageFilter, this.location = locationFilter,
        this.zipcode = zipFilter, this.firstName = firstNameFilter, this.lastName = lastNameFilter,
        this.branch = branchFilter,
        (resRecordsData) => {
            this.records = resRecordsData;
         });
 }, 200);

public onFilterReceived(values)
{
    let languageFilter = [];
    let locationFilter = [];
    let zipFilter = [];
    let firstNameFilter = [];
    let lastNameFilter = [];
    let branchFilter = [];

     this.route.params.subscribe(
         (params: any) => {
             this.page = params['page'];
             this.pn_zip_e = params['pn_zip.e'];
             this.pn_firstName_e = params['pn_firstName.e'];
             this.pn_lastName_e = params['pn_lastName.e'];
             this.pn_location_e = params['pn_location.e'];
             this.pn_branch_e = params['pn_branch.e'];
             this.pn_language_e = params['pn_language.e'];
         }
     );

     this.pn_language_e === "1" ? languageFilter = values['language'] : languageFilter = [];
     this.pn_location_e === "1" ? locationFilter = values['location'] : locationFilter = [];
     this.pn_zip_e === "1" ? zipFilter = values['zip'] : zipFilter = [];
     this.pn_firstName_e === "1" ? firstNameFilter = values['firstName'] : firstNameFilter = [];
     this.pn_lastName_e === "1" ? lastNameFilter = values['lastName'] : lastNameFilter = [];
     this.pn_branch_e === "1" ? branchFilter = values['branch'] : branchFilter = [];

     this.sendRequest(languageFilter, locationFilter, zipFilter, firstNameFilter, lastNameFilter, branchFilter);
};

The getByFilters() function being called from my filtersService look like this:

public getByFilters(page, pagesize, stage?, language?, location?, zipcode?, firstName?, lastName?, branch?, fn?: any)
{
    return this.apiService.get({
      req: this.strReq, reqArgs: { page, pagesize, stage, language, location, zipcode, firstName, lastName, branch }, callback: fn });
}

And this in turn calls a GET request in our central request controller service (apiService) which looks like this:

public get(args: {
    req: string,
    reqArgs?: any,
    reqBody?: any,
    callback?: IRequestCallback
}): Observable<Response>
{
    args['reqType'] = 'get';
    return this.runHttpRequest(args);
}

Once I get my response I assign it to "this.records", and then use that in my view to iterate over "this.records.data" -- which is an array, to print my records to the screen.

So from my above code, this line is where I get the response and assign it to "this.records":

  this.records = resRecordsData;

So, my first question is, how do I determine if I have an RxJS observable here - and then how do I use an operator like switchMap() to handle cancelling previous filter requests?

I tried this, but it's not working. My guess is the syntax is incorrect:

private sendRequest = _.debounce((languageFilter, locationFilter, zipFilter, firstNameFilter, lastNameFilter, branchFilter) =>
{
    this.streamFiltersService.getByFilters(
        this.page - 1, this.pagesize, this.currentStage, this.language = languageFilter, this.location = locationFilter,
        this.zipcode = zipFilter, this.firstName = firstNameFilter, this.lastName = lastNameFilter,
        this.branch = branchFilter,
         (resRecordsData) => {
            resRecordsData.switchMap((res) => {
                this.records = res;
            });
            console.log('records: ', this.records);
         });
 }, 200);

First off, to make sure I'm barking up the right tree, I'd like to have a way to determine if the response I have here actually is an RxJS observable. And, if not, find a way to convert it to one, so I can then use switchMap() on it.

Answer 1

i have created real case for switchMap and two input with reactiveForm. This can be adapt with fromEvent(inputElement,'change') instead of this.inputElement.valueChanges.

// We want combine of all lastest values.
combineLatest(
  // We merge default name with observable of name input.
  merge(
    of('default'),
    // From input we want to be inform after 200ms debounce and if they have change.
    this.name.valueChanges.pipe(debounceTime(200), distinctUntilChanged())
    ),
  // We merge default choice with observable of choice input.
  merge(
    of('value 1'),
    this.choice.valueChanges.pipe(debounceTime(200), distinctUntilChanged())

  )
).pipe(map(e => {
  // We construct the request payload
  return {
    name: e[0],
    choice: e[1]
  }
  // Ignore combined default value, ask api and subscribe to answer.
})).pipe(skip(1), switchMap(e => this.myApi(e))).subscribe(console.log);

To be sure to understand each step, i highly recommand you to split it by composed variable, and console.log each subscribed stream. Example :

Input observable

this.name.valueChanges.pipe(debounceTime(200), distinctUntilChanged()).subscribe(console.log)

will output you current input value after debounce of 200ms and if value have change compare to previous emitted one.

Input observable with default value

merge(
    of('default'),
    this.name.valueChanges.pipe(debounceTime(200), distinctUntilChanged())
),

We merge immediate default value with future value emitted on input stream.

CombineLatest

from previous composed merge, we want to combine latest emitted value from each input as single stream.

map

.pipe(map(e => {
  // We construct the request payload
  return {
    name: e[0],
    choice: e[1]
  }
  // Ignore combined default value, ask api and subscribe to answer.
})

because combineLatest will create array of latest emitted stream provided as parameter.. We want to map this array to the real payload object for your api.

switchMap

switchMap(e => this.myApi(e)))

You have your payload (produce by map describe previously), you transform it to new observable. Now you can simply subscribe and magically you will have answer of your API base on your collection of inputs values, and automatically cancel previous request which are not relevant anymore.

live sample

---

To be base on your route Params, you can do something like this.

  ngOnInit() {
    // From url params change, we map output with default value.
    const routeParams$ = this.route.params.pipe(map(params => {
      return {
        user: '',
        choice: '',
        ...params
      };
    }));
    // Uncomment me to see it in action.
    // routeParams$.subscribe(console.log);
    // Transform this payload to http request and subscribe to it.
    routeParams$.pipe(switchMap(e => this.myApi(e))).subscribe(console.log);
  }

Live sample