RxJS add catch() to an Observable that's been shared

Question

I have the following code I'm using in my Angular (4.0) application during $http calls.

return this.httpComponent.post(serviceUrl, request, args)
            .map((res: any) => res.json() as R)
            .catch((error: any) => Observable.throw(error.json().error || error.json().errorMessage || 'Server error'));

After testing this I realized multiple subscriptions were triggering the request multiple times. Thanks to this post: Angular2 http.post gets executed twice I found out I needed to share() the result.

This works to get rid of the multiple calls but now it seems my catch() method isn't being hit. I want my catch() to throw the error.

I tried the two options below but they did not work:

return this.httpComponent.post(serviceUrl, request, args)
    .map((res: any) => res.json() as R)
    .share()
    .catch((error: any) => Observable.throw(error.json().error || error.json().errorMessage || 'Server error'));

return this.httpComponent.post(serviceUrl, request, args)
    .map((res: any) => res.json() as R)
    .catch((error: any) => Observable.throw(error.json().error || error.json().errorMessage || 'Server error') 
    .share()); //This doesn't make sense since my catch() isn't returning an Observable

Anyone know how I can share() and catch(throw...) at the same time?

score 2 · Answer 1 · answered Jul 19 '17 at 13:18

As mentioned in my comment to @cartant 's answer, if any subscriber doesn't have an error handler (ie: doesn't care about any error scenarios) an exception is thrown and all subsequent subscribers are never informed of the original error.

This seems like a design flaw in my opinion. Here is the example (Copied from @cartant 's answer)

const source = Rx.Observable
  .interval(500)
  .map((value) => {
    if (value === 2) {
      throw new Error("Boom!");
    }
    console.log(`value: ${value}`)
    return value;
  })
  .catch((error) => Rx.Observable.throw(new Error(`Re-thrown ${error.message}`)))
  .share();

source.subscribe(
  (value) => console.log(`subscription 1: ${value}`)
);
source.subscribe(
  (value) => console.log(`subscription 2: ${value}`),
  (error) => console.log(`subscription 2: ${error}`)
);

.as-console-wrapper { max-height: 100% !important; top: 0; }

<script src="https://unpkg.com/rxjs/bundles/Rx.min.js"></script>

FYI, it seems there is an open issue for this: https://github.com/ReactiveX/rxjs/issues/2180 - so you're not the only person to consider it a flaw. — cartant, Jul 19 '17 at 13:33
Thanks for the link. I definitely think this behavior should be changed. — Sal, Jul 19 '17 at 14:52

cartant · Accepted Answer · 2018-04-25T23:05:46.183

The second option you've included in your question attaches the catch to the Observable.throw. If that is corrected, you should see the behaviour you are expecting:

const source = Rx.Observable
  .interval(500)
  .map((value) => {
    if (value === 2) {
      throw new Error("Boom!");
    }
    console.log(`value: ${value}`)
    return value;
  })
  .catch((error) => Rx.Observable.throw(new Error(`Re-thrown ${error.message}`)))
  .share();

source.subscribe(
  (value) => console.log(`subscription 1: ${value}`),
  (error) => console.log(`subscription 1: ${error}`)
);
source.subscribe(
  (value) => console.log(`subscription 2: ${value}`),
  (error) => console.log(`subscription 2: ${error}`)
);

.as-console-wrapper { max-height: 100% !important; top: 0; }

<script src="https://unpkg.com/rxjs@5/bundles/Rx.min.js"></script>

The code in my example is actually a typo. I wasn't trying to attach the **share()** to the **throw()**. But your code snippet helped me figure out the problem so I'm accepting your answer. As for my problem, it turns out one of the earliest subscribers didn't have an error handler. So, and this seems like a design flaw imo, all subsequent subscribers would never get the error. — Sal, Jul 19 '17 at 13:15

RxJS add catch() to an Observable that's been shared

2 Answers2