Как сделать очередь http get/post запросов Angular 2+?

Question

[sdgroup14]

У меня есть к примеру 25 разных post или get.... Как мне сделать так, чтобы 2рой запрос не начался пока 1 не завершится... 1 завершился начался 2. пока 2рой запрос не завершится не начинать 3тий и т.д. У меня обязательно нужно сделать такое условие. Во первых из за рефреш токена... А во вторых логика предполагает такое. Ниже скинул шаблон http шаблон error и пример get/post у меня еще таких штук 40 так что шаблонная функция обязательно нужна...

devicesTypes = new Subject<any>();
  usersAccountList = new Subject<any>();
  userDevice = new Subject<any>();
  userDeviceNotification = new Subject<any>();
  allCategories = new Subject<any>();

// и т.д.....

  httpGetTemplate(param_url?, target?) {
    const httpHandler = () => {
      const $params = param_url;
      const $target = target;
      const options = {
        headers: new HttpHeaders().set('Authorization', 'Bearer ' + localStorage.getItem('token'))
      };
      return this.http.get(this.apiURL + $params, options).subscribe((_data: any) => {
        // console.log('new http :', _data);
        $target.next(_data.result);
      }, _error => {
        // console.log('new _error :', _error);
        this.errorsHandler(_error, httpHandler);
      });
    };
    httpHandler();
  }

  errorsHandler(_err, _httpHandler) {
    if (_err.status === 401) {
      const optionsSimple = {
        headers: new HttpHeaders().set('Content-Type', 'application/x-www-form-urlencoded')
      };

      const refresh = new URLSearchParams();
      refresh.set('refresh_token', localStorage.getItem('refresh_token'));
      this.http.post('http://185.156.41.135/api/web/v1/users/refresh-token', refresh.toString(), optionsSimple).subscribe((data: any) => {
        localStorage.setItem('token', data.result.auth_token);
        localStorage.setItem('refresh_token', data.result.refresh_token);
        _httpHandler();

      }, __err => {
        if (__err.status === 400) {
          localStorage.removeItem('token');
          localStorage.removeItem('refresh_token');
          this.router.navigate(['/login']);
        }
      });
    } else if (_err.status === 400) {
      // localStorage.removeItem('token');
      // localStorage.removeItem('refresh_token');
      // this.router.navigate(['/login']);
    }
  }

  getleftWidgetCategories() {
    return this.httpGetTemplate('/device-info/devices', this.allCategories);
  }

  getDevicesTypes() {
    return this.httpGetTemplate('/devices/get-device-models', this.devicesTypes);
  }


  getUsersAccountList() {
    return this.httpGetTemplate('/users/list', this.usersAccountList);
  }

  getUserDevice() {
    return this.httpGetTemplate('/devices/get-device-user', this.userDevice);
  }

  getUserDeviceNotification(id) {
    return this.httpGetTemplate('/devices/settings-event?id=' + id, this.userDeviceNotification);
  }

  getDeviceUserAcccess(user_id, device_id) {
    return this.httpGetTemplate('/devices/get-device-user-acccess?user_id=' + user_id + '&device_id=' + device_id, this.deviceUserAcccess);
  }

Answer 1

[Антон Швец]

Как мне сделать так, чтобы 2рой запрос не начался пока 1 не завершится... 1 завершился начался 2. пока 2рой запрос не завершится не начинать 3тий и т.д.

switchMap
или concatMap
зависит от конкретики вашей задачи

Comments

[sdgroup14]

у меня есть виджет A по клику на итем в виджете A идет запись в базу(это первый запрос). После клика на итем в виджете B должен пойти запрос по данным кликнутого Итема(2 запрос)....
Таких цепочек у меня довольно много и вот как то мне нужно это поместить все в очередь... Бывают даже цепочки А В С.... В данный иногда очень большой пендинг в нетворках... И у меня бывает запрашивает по 3 раза token.... и первый решфреш токен забрал и сработал норм... а 2 и 3 запрос прислали 400 ошибку и я вылетел с приложения на логинку..... Для того чтоб это избежать... мне нужно все ставить в очередь.... И как бы по доке я это должен сделать.... Я думал что я такие штуки должен бы пушить в какойто Subject общий для запросов и в котором выполняется первый запрос начинается 2...

[Антон Швец]

sdgroup14, набросал такое, думаю можно как-то изящнее, но с утра плохо думается

import { Subject, of } from 'rxjs';
import { delay, delayWhen } from 'rxjs/operators';

const triggerA = new Subject();

const itemA = of('itemA').pipe(delay(1000));
const itemB = of('itemB').pipe(delayWhen(() => triggerA));

itemA.subscribe(r => {
  console.log(r);
  triggerA.next()
});
itemB.subscribe(r => {
  console.log(r)
});

[Антон Швец]

sdgroup14, а вот, если надо выполнить последовательно, но одним скопом.

import { Subject, of } from 'rxjs';
import { delay, concat } from 'rxjs/operators';

const itemA = of('itemA').pipe(delay(1000));
const itemB = of('itemB');

itemA.pipe(concat(itemB)).subscribe(r => {
  console.log(r)
});

[sdgroup14]

ну он же просто секунду будет ждать? И тем более Я не знаю Что запустится После A.... может вызваться не B а F.... а при клике на F может вызваться как и D так и Q.. Все завист он параметров какие выбраны. Мне нужно какойто массив []. в который я буду пушить http post или get и опять же рандомный... Опять же этот массив должен быть в сервисе... потому что вызовы с разных компонентов.... по клику пушится в массив и вызывается если он пустой.... если не пустой то значит поместить в конец и поставить в очередь на вызов

[Антон Швец]

sdgroup14, секунду это я поставил, просто чтобы отсрочить выполнение обсервабла itemA, т.е. имитировать асинхронность.
А itemB выполнится только если сработал triggerA, если нет, он будет ждать. На этот триггер вы можете повесить любую логику. В том числе и слежение за вашим массивом.

[Антон Швец]

sdgroup14, ваша очередь

import { Subject, of } from 'rxjs';
import { delay } from 'rxjs/operators';

const queue$ = new Subject();
const clear$ = new Subject();
let queue = [];
queue$.subscribe(q => queue = q);

const addToQueue = (...args) => queue$.next(queue.concat(args));

// cycle
queue$.subscribe(() => {
  if (!queue.length) {
    clear$.next();
  } else {
    queue.shift().subscribe(res => {
      console.log(new Date(), res);  // logic
      queue$.next(queue);
    });
  }
});


addToQueue(
  of('itemA').pipe(delay(10)),
  of('itemB').pipe(delay(1000)),
  of('itemC').pipe(delay(500)),
  of('itemD').pipe(delay(2000)),
);

[sdgroup14]

Антон, Cпасибо начинаю понимать!

[Антон Швец]

sdgroup14, заметьте, что в этом случае у всех одинаковый subscribe
Индивидуальную логику можно выполнить заранее в .pipe(tap(res => /* here */))

[Антон Швец]

sdgroup14, апдейтну, а то некорректно работает. Ну чисто как задачку для себя

import { Subject, of, Observable } from 'rxjs';
import { delay } from 'rxjs/operators';

export class Queue {
  private additions: Observable<any>[] = [];
  private queue$ = new Subject<Observable<any>[]>();
  private clear$ = new Subject<void>();
  private queue: Observable<any>[] = [];

  constructor() {
    this.queue$.subscribe(q => this.queue = q);

    this.queue$.subscribe(() => {
      this.queue = this.queue.concat(this.additions);
      this.additions = [];
      if (!this.queue.length) {
        this.clear$.next();
      } else {
        this.queue.shift().subscribe(res => {
          console.log(res, '   to do:', this.queue.length + this.additions.length);
          this.queue$.next(this.queue );
        });
      }
    });

    this.clear$.subscribe(() => {
      console.log('done');
    })
  }

  public addTask (...args) {
    this.additions = this.additions.concat(args);
    console.log('-- added:', this.additions.length, 'total: ', this.queue.length + this.additions.length);
    if (!this.queue.length) {
      this.queue$.next(this.queue);
    }
  }

  public done() {
    return this.clear$.asObservable();
  }

  public nextWill() {
    return this.queue[0];
  }

  public count() {
    return this.queue.length;
  }
}

const queue = new Queue();

queue.addTask(
  of('itemA').pipe(delay(1000)),
  of('itemB').pipe(delay(1000)),
  of('itemC').pipe(delay(1000)),
  of('itemD').pipe(delay(10)),
);
queue.addTask(
  of('itemE').pipe(delay(1000)),
  of('itemF').pipe(delay(1000)),
  of('itemG').pipe(delay(1000)),
  of('itemH').pipe(delay(10)),
);

setTimeout(() => {
  queue.addTask(of('itemI').pipe(delay(1000)));
}, 10000);

[sdgroup14]

спасибо вам огромное

[sdgroup14]

Антон, а операторы of? это из новых операторов? То я уже запутался с новым и старым. of, pipe никогда не пользовался...

[Антон Швец]

sdgroup14, of был давно, это порождающий оператор Observable.of
Ничего не изменилось, просто теперь их можно импортировать отдельно, а pipe это просто обертка для старых операторов изменения, которая тоже позволяет их импортировать как отдельные функции.

[sdgroup14]

спасибо еще раз :)