Issue with Observable subscription not receiving notifications

Question

Issue with Observable subscription not receiving notifications

I am currently working with an Observable to convert a promise into a subscription. This leads to a collection that requires iteration in order to call an HTTP Service on each element. I am using forkJoin to wait for all the calls to finish before proceeding, but unfortunately, my subscription is not triggering. Is there something crucial that I may be overlooking?

Observable.fromPromise(this.users.getElements()).subscribe(results => {
  Observable.forkJoin(
    results.map(
      aUser => this.HttpService.submitUser(aUser).subscribe(
        results => {
          this.progress += 1;
        },
        err => {
          this.progress += 1;
          this.handleError(<any>err);
        })
    ).subscribe(
      //the code never reaches these calls after completing all service calls
      data => {
        debugger;
        console.log(data);
        this.reset();
      },
      err => {
        debugger;
        console.log(err);
        this.reset();
      }
    ));
});

angular typescript rxjs rxjs5

Answer 1

Answer №1

One important point is that you do not need to subscribe to each Observable individually when using the forkJoin() method. The operator handles this internally.

If you wish to receive a notification when each Observable completes, you can achieve this by using the

.do(undefined, undefined, () => {...})

callback.

let streams = [
  Observable.of(42).do(undefined, undefined, () => console.log('done')),
  Observable.of('a').delay(100).do(undefined, undefined, () => console.log('done')),
  Observable.of(true).do(undefined, undefined, () => console.log('done')),
];

Observable.forkJoin(streams)
  .subscribe(data => console.log(data));

The output in the console will be:

done
done
done
[ 42, 'a', true ]

Additionally, there is a .finally() operator available. Nevertheless, it functions differently from the .do() method.

UPDATE:

In case any of the source Observables encounter an error, the forkJoin() operator propagates that error (leading to failure).
Thus, individual errors from each source Observable should be handled separately (e.g., using the catch() operator).

let streams = [
  Observable.throw(new Error())
    .catch(() => Observable.of('error caught 1'))
    .do(undefined, undefined, () => console.log('done 1')),

  Observable.of('a')
    .delay(100).catch(() => Observable.of('error caught 2'))
    .do(undefined, undefined, () => console.log('done 2')),

  Observable.of(true)
    .catch(() => Observable.of('error caught 3'))
    .do(undefined, undefined, () => console.log('done 3')),
];

Observable.forkJoin(streams)
  .subscribe(data => console.log(data));

This will display the following results:

done 1
done 3
done 2
[ 'error caught 1', 'a', true ]

Answer 2

One important point is that you do not need to subscribe to each Observable individually when using the forkJoin() method. The operator handles this internally.

If you wish to receive a notification when each Observable completes, you can achieve this by using the

.do(undefined, undefined, () => {...})

callback.

let streams = [
  Observable.of(42).do(undefined, undefined, () => console.log('done')),
  Observable.of('a').delay(100).do(undefined, undefined, () => console.log('done')),
  Observable.of(true).do(undefined, undefined, () => console.log('done')),
];

Observable.forkJoin(streams)
  .subscribe(data => console.log(data));

The output in the console will be:

done
done
done
[ 42, 'a', true ]

Additionally, there is a .finally() operator available. Nevertheless, it functions differently from the .do() method.

UPDATE:

In case any of the source Observables encounter an error, the forkJoin() operator propagates that error (leading to failure).
Thus, individual errors from each source Observable should be handled separately (e.g., using the catch() operator).

let streams = [
  Observable.throw(new Error())
    .catch(() => Observable.of('error caught 1'))
    .do(undefined, undefined, () => console.log('done 1')),

  Observable.of('a')
    .delay(100).catch(() => Observable.of('error caught 2'))
    .do(undefined, undefined, () => console.log('done 2')),

  Observable.of(true)
    .catch(() => Observable.of('error caught 3'))
    .do(undefined, undefined, () => console.log('done 3')),
];

Observable.forkJoin(streams)
  .subscribe(data => console.log(data));

This will display the following results:

done 1
done 3
done 2
[ 'error caught 1', 'a', true ]

Answer 3

Answer №2

There might be no need to subscribe on the map in this scenario.

Observable.fromPromise(this.users.getElements()).subscribe(results => {
  Observable.forkJoin(
    results.map(
      aUser => this.HttpService.submitUser(aUser))
    ).subscribe(
      //The calls after all service calls are never reached
      data => {
        debugger;
        console.log(data);
        this.reset();
      },
      err => {
        debugger;
        console.log(err);
        this.reset();
      }
    ));
});

This rxjs example shows that they don't directly subscribe to individual observables - ForkJoin kicks them off, then waits for all of them to return (in your subscribe block).

Edit:

You can find the source code for forkjoin here:

https://github.com/Reactive-Extensions/RxJS/blob/master/src/core/linq/observable/forkjoin.js

It seems like there are no hooks available to track when each one finishes. Perhaps a better approach would be to individually subscribe to each mapped observable, increment the UI counting bar variable, and have a completion check to utilize the data effectively.

Answer 4

There might be no need to subscribe on the map in this scenario.

Observable.fromPromise(this.users.getElements()).subscribe(results => {
  Observable.forkJoin(
    results.map(
      aUser => this.HttpService.submitUser(aUser))
    ).subscribe(
      //The calls after all service calls are never reached
      data => {
        debugger;
        console.log(data);
        this.reset();
      },
      err => {
        debugger;
        console.log(err);
        this.reset();
      }
    ));
});

This rxjs example shows that they don't directly subscribe to individual observables - ForkJoin kicks them off, then waits for all of them to return (in your subscribe block).

Edit:

You can find the source code for forkjoin here:

https://github.com/Reactive-Extensions/RxJS/blob/master/src/core/linq/observable/forkjoin.js

It seems like there are no hooks available to track when each one finishes. Perhaps a better approach would be to individually subscribe to each mapped observable, increment the UI counting bar variable, and have a completion check to utilize the data effectively.

Issue with Observable subscription not receiving notifications

Answer №1

Answer №2

Similar questions

Developing a login feature in Angular 2 using Spring Security

Tips for inputting transition properties in Material UI Popper

The absence of the function crypto.createPrivateKey is causing issues in a next.js application

TypeScript: a sequence of symbols representing a particular <type>

Google Material Icons are failing to render in Angular application

Using Javascript or ES6, you can compare a nested array object with another array of elements and generate a new array based on

"TypeScript error: Arrays in interfaces are found to be undefined and not compatible

Using `it` with accessing class members

Creating a new object in a Redux selector and returning it can be achieved by following these steps

Start up a server using Angular along with Node.js and Express framework

Unable to locate necessary assets, Electron encounters a resource deficiency

The 'Subscription' type does not contain the properties _isScalar, source, operator, lift, and several others that are found in the 'Observable<any>' type

Simplify an array in Javascript

Utilizing ES6, accessing the first element of an array of objects

Using React with TypeScript to implement a nested map function

Angular 4 Reactive Forms: How to Bind Data to Multiple Checkboxes

Managing animations with multiple components in Angular 2+

"Enhance your coding experience in VS Code with intelligent auto-completion for JavaScript files using type definitions that support import

typescript set x and y values to specific coordinates

Generate an array by filtering out null properties from a Javascript/Typescript object