Notify when all observables are completed

Question

I have an array of observables. Each observable is a file which needs to be uploaded via HTTP request.

I want to do two things:

1. For each uploaded file I have to emit event which will be handled in parent component.
2. When all files are uploaded I have return notify that all observable are completed.

I started with this code:

this.isUploading$ = Observable.of(true);

const source$ = Observable.from(files)
    .concatMap(file => this.uploadSingleFile(file));

source$.subscribe(data => this.onSuccess.emit(data));

Observable isUploading$ represent that upload process started and it is consumed in HTML. I want to display spinner based on the state of isUploading$.

Observable source$ represents file uploads operations because method this.uploadSingleFile(file) with return Observable<Response>.

This code will upload all files and execute this.onSuccess.emit(data) on each completed observable.

The question is: How I can set this.isUploading$ to false when everything is done?

UPDATE:

I would like to achieve something like this but without assigning variable in onCompleted function. Example:

source$.subscribe(data => this.onSuccess.emit(data),
                  err => console.log(err),
                  () => this.isUploading = false );

I want to retrieve two observables and subscribe to them later whenever I want.

Answer 1

If you want to run all of these observables concurrently, concatMap will not do what you want. You can do these two things with the do and forkJoin operators.

forkJoin is basically Promise.all for observables.

do is a way to tap into the values being passed through and observable so you can "do" something with them (a side-effect).

So:

const arrayOfObservables = [observable1, observable2, observable3];

Observable.forkJoin(
  arrayOfObservables.map((obs, i) => obs.do({
    next(value) { console.log(`Observable ${i} emits: ${value}`); },
    complete() { console.log(`Observable ${i} is complete`); }
  }))
)
.subscribe(values => console.log('everything done with', values))

Answer 2

Why do you need isUploading to be an observable? I'd do it like this:

this.isUploading = true;
const source$ = Observable.from(files)
    .concatMap(file => this.uploadSingleFile(file));

source$.subscribe(data => this.onSuccess.emit(data),
                  err => console.log(err),
                  () => this.isUploading = false );

Answer 3

You could simply use a BehaviorSubject for it:

public isUploading$ = new BehaviorSubject<boolean>(true);

...

this.isUploading$.next(false);
this.isUploading$.complete();

And use forkJoin operator to wait for all to complete:

const allSources$ = Observable.forkJoin(source$);
allSources$.subscribe(data => {
    this.isUploading$.next(false);
    this.isUploading$.complete();
}

This would give you the desired result.