RXJS queue up multiple http requests and send when x conditions happens

Question

Just started to use rxjs few weeks ago, I think I'm getting most of it right, but my code seems to be not that efficient and prone to bugs I haven't run into yet!

I need to be able to send http requests my app is online(1) and when the user is authenticated(2) and when there's anything to send(3). I'm pretty sure there's a smart way to combine those 3 conditions the "rxjs" way :)

SuperHttpRequest

class SuperHttpRequest extends HttpRequest<any> {

  _id: string; // Used for localStorage

  constructor(method: string, url: string, body: any, _id?: string) {
    super(method, url, body);
    this._id = _id || UUID.UUID();
  }
}

HttpRequestHandler

@Injectable()
class HttpRequestHandler extends ReplaySubject<SuperHttpRequest> {

  nbItemsInQueue$: ReplaySubject<Number>; // Needed by a component (sync progress), would be better to use the combined http request time
  private requests: SuperHttpRequest[];
  private requestsSubcription: Subscription;

  constructor(
    private network: Network, // Ionic service
    private http: HttpClient,
    private localStore: LocalStore<SuperHttpRequest[]>,
    private authService: AuthenticationService) {

    super();

    this.requests = [];
    this.nbItemsInQueue$ = new ReplaySubject(this.requests.length);

    this.localStore.get('http-request')
      .then(requests => {
        if (requests) {
          requests.forEach(request => {
            this.next(new SuperHttpRequest(request.method, request.url, request.body, request._id));
          });
        }
      });

    this.nbItemsInQueue$
      .subscribe(() => this.localStore.set('http-request', this.requests));

    this.authService.currentUser$
      .subscribe(user => {
        if (!user) { if (this.requestsSubcription) { this.requestsSubcription.unsubscribe(); } }
        else {
          this.requestsSubcription = this
            .takeWhile(() => {
              if (navigator.onLine) { return true; }
              else { throw Error(); }
            })
            .retryWhen(() => this.network.onConnect())
            .subscribe(request => this.handleHttpRequest(user, request));
        }
      });
  }

  private handleHttpRequest(user: Credentials, request: SuperHttpRequest) {
    this.http.request(request.clone({ setHeaders: { Authorization: `Bearer ${user.token}` } }))
      .retryWhen(error => {
        return error
          .mergeMap((error: any) => {   
            if (error.status === 503) {
              return of(error.status).delay(2000)
            }
            return Observable.throw({ error: 'No retry' });
          })
          .take(5)
          .concat(Observable.throw({ error: 'Sorry, there was an error (after 5 retries)' }));
      })
      .subscribe(res => {
        if (res instanceof HttpResponse) {
          remove(this.requests, (req) => req._id === request._id);
          this.nbItemsInQueue$.next(this.requests.length);
        }
      });
  }

  next(request: SuperHttpRequest) {
    this.requests.push(request);
    this.nbItemsInQueue$.next(this.requests.length);
    super.next(request);
  }

}

Answer 1

//Maintain Queue of http Observable
const quequeHTTP = [];

//Dummy Http Request
getData(){
   return this.http.get('./api');
}

addInQueuqe(observable:Observable<any>){
     quequeHTTP.push(observable);
}

//Dummy addition 4 request
addInQueuqe(this.getData());
addInQueuqe(this.getData());
addInQueuqe(this.getData());
addInQueuqe(this.getData());

eventWhenXHappen(){
  Observable.forkJoin(quequeHTTP).subscribe((resArr)=>{
     console.log('When all request succesfully executed ');
},(err)=>{
  console.log('When error occur');
})


}