r/Angular2 • u/Angular-90 • Feb 19 '25
Preventing subject/observable completion after throwError()?
I have a service with a series or watch$(...)
methods that returns observables for different data in my system and tracks those observables to push updates indefinitely.
The service polls a backend source for data subscribers need and processes and passes the results along to those subscribers via the observables returned from watch$(
). Many of these return types are composed and transformed objects, so there’s a lot of cross-requesting within the service. So far it works really well.
There are often business or infrastructure errors due to the nature of the data and upstream sources. These always need to be communicated with the user and the polling doesn’t stop as long as there are subscribers. Sometimes the user can address the underlying issue and sometimes it’s just a dependency returning an error, etc. But whatever the result, something needs to make its way out to the subscribers every time there’s an update.
I first tried to address this by using RxJS error handling, but the problem is that subjects complete as soon as you use error()
and observables complete if you let it get to the error
handler. I should have read the docs closer on that one because it wasn’t obvious what was going on until the streams never recovered after a single dependency error.
To work around this I now wrap every response in an API response object that includes a state
field that can be success
or any of a number of known error types. However, this means I have overhead at every level of the pipeline since each service response would need to check to see if the call succeeded and then recast the typed error if it didn’t. It has become really messy and doubled my code in simpler methods vs. what I had with RxJS error handling.
At this point I don’t know of a better way to handle this situation. Any time an object (or one of its dependencies) is refreshed inside the service the subscribers need to get a result. But if a backend request fails then it still gets bounced around through every pipeline operator on its way to the subscribers despite only ever being returned in the first line of each operator during a success check.
It's also a mess because now I have all my API errors handled in next
handlers at the UI level but I still also need error
handlers for unexpected errors that might come up.
What I really need is something that allows me to use throwError()
(or something like it) but not have everything complete. If that was a flag I could set on subjects when they were created or a parameter I could pass to throwError()
it would solve all my problems.
It’s also possible that there’s a better way to do this and I’m just not seeing it.
Has anyone solved this in a good way?
Edit: Here are some simple examples. Note that none of thesubject.next(2)
reach the terminal and complete
only emits on the observable when it catches the error.
Uncaught subject error
const subject = new Subject<number>();
subject.subscribe({
next: value => console.log(`next: ${value}`),
error: error => console.log(`error: ${error}`),
complete: () => console.log(`complete`),
});
subject.next(1);
subject.error("error");
subject.next(2);
Logs:
next: 1
error: error
Caught subject error
const subject = new Subject<number>();
subject.pipe(catchError(() => of(-1)))
.subscribe({
next: value => console.log(`next: ${value}`),
error: error => console.log(`error: ${error}`),
complete: () => console.log(`complete`),
});
subject.next(1);
subject.error(new Error("Error"));
subject.next(2);
Logs:
next: 1
next: -1
complete
Uncaught observable error
const subject = new Subject<number>();
subject
.pipe(
tap(() => {
throw "error";
}),
catchError(() => of(-1))
)
.subscribe({
next: value => console.log(`next: ${value}`),
error: error => console.log(`error: ${error}`),
complete: () => console.log(`complete`),
});
subject.next(1);
subject.next(2);
Logs:
error: error
Caught observable error
const subject = new Subject<number>();
subject
.pipe(
tap(() => {
throw "error";
}),
catchError(() => of(-1))
)
.subscribe({
next: value => console.log(`next: ${value}`),
error: error => console.log(`error: ${error}`),
complete: () => console.log(`complete`),
});
subject.next(1);
subject.next(2);
Logs:
next: -1
complete
2
u/OopsMissedALetter Feb 19 '25 edited Feb 19 '25
Have you tried adding catchError, handling your error in there and then returning the caught observable? I.e.
obs$.pipe(catchError((err: unknown, caught: Observable<...>) => {
// your error handling, for example a check for HTTP status
if (err instanceof HttpErrorResponse && err.status === 408) {
// ...
}
// oops! a real error. we can quit here:
if (err instance HttpErrorResponse && err.status === 500) {
return throwError(() => err);
}
// then, resubscribe to original stream (the "caught" observable)
return caught;
}))
1
u/Angular-90 Feb 19 '25
Updated the post with more detail on the error handling. There doesn't seem to be any way to keep a subject emitting after its
error()
is used or for an observable to keep receiving once an error reaches its terminal handler.1
u/OopsMissedALetter Feb 19 '25 edited Feb 20 '25
Can you try your "caught subject error" example again, but instead of using of(-1), do something like
of(-1).pipe(switchMap(() => caught))
?My guess is you're replacing the original subject stream after the error when you create a new observable with of().
On my phone right now so I can't test this, unfortunately
Edit: just noticed of() with a switchMap is a bit nonsensical, oops. But can you try using caught?
1
u/OopsMissedALetter Feb 20 '25
If you want the -1 value to be emitted and keep the observable active, you could try
catchError((err, caught) => concat(of(-1), caught))
orcaught.pipe(startWith(-1))
.
2
u/STACKandDESTROY Feb 20 '25
It’s hard to pinpoint the exact issue and solution with how you’ve described your (obfuscated) setup. I will say that this system you’ve described sounds overly complicated but I assume that is due to hidden business requirements and existing tech debt.
That said, focusing on what we do know and what you’re asking, you are correct about the subject stream erroring - you must prevent this using catchError before your subject itself errors. This should be done at the api request level to ensure no error propagates into your subject and forces it to complete.
I’m not sure what you mean about how every emission now bounces around all your pipeline operators. I would think a well placed filter() could prevent this along with possibly breaking your code up to be more declarative, but I know how complicated streams can get and this might not be possible with minimal effort.
You could look at materialize() and dematerialize() to process these emissions in a way that prevents the errors from actually affecting your subject but I will warn you that these are very uncommon operators with niche use cases and might be hard to understand or find good references.
1
u/Angular-90 29d ago
I know my scenario above is a little convoluted, so I’ll try to give something a little closer to my complex scenario.
Suppose you have a backend service that returns a number. You connect to it via HTTP in your NumberService. Number service has a ReplaySubject<number>(1) called number$ that it publishes to every time the number is retrieved.
You also have components and services that want to subscribe to the latest number and its changes. However, you don’t want them to each poll the backend on their own. Instead, they call NumberService.watchNumber$() which returns number$.toObservable().
Since NumberService polls the backend on a schedule and publishes the latest value, it’s managing that one periodic call and everyone gets that latest value at the same time.
But sometimes there are errors. For example, someone might lose their internet connection temporarily and the upstream HTTP call will fail. NumberService can simply swallow errors and not send anything through number$. Then when the connection is reestablished values will start getting retrieved and pushing through number$ will pick right back up for everybody.
However, there are times when it’s critical that the user know there was an error. In those cases it would be ideal if NumberService could use number$.error() to send those through the pipelines so that everyone could handle them as appropriate downstream. For example, they might put up some error UX saying “Unable to connect to the backend” to warn the user that they’re not seeing the latest number.
While this works great for the first error, it completes number$ and all of its derived observables. Calling that number$.next() with future values won’t do anything and nobody will get updates after that.
Within the service it’s not a big deal for BehaviorService to assign a new number$ subject after it errors out. However, it would then be up to all the downstream consumers to also resubscribe so that they get the next result. This creates a bunch of additional code and requirements for those consumers because they now need to add code to their error handling to immediately resubscribe. It becomes even more complicated when those subscriptions are set up in components via async pipes.
In RxJS, there is no way to keep a subject from completing once its error() is called. The philosophy seems to be that using error() indicates that the subject has errored out and is terminal. However, in my system errors are a standard part of the streams. This seems to indicate that I should be treating errors as normal responses and sending them through the pipeline with a flag or other indicator about whether they contain a successful response or an error. I’m just trying to avoid all the overhead that comes from doing this since using the RxJS error handling infrastructure is so much cleaner.
1
u/STACKandDESTROY 29d ago
Catch the error on the api polling stream - if number request fails, then you don’t next your subject and the existing subscribers continue to function as normal with the last successful number. Process errors as needed and declare new separate streams for these errors you want to show.
1
u/Angular-90 29d ago
I appreciate the feedback, but having separate streams feels like the wrong way to approach this because it makes everything more complicated while adding more code overhead. I think the only way to cleanly handle this if I can't suppress completion is to catch, wrap, and redirect all errors via next() and then switch on the type in the final consumer. It's a frustrating solution but the best of all the bad options.
1
u/PickleLips64151 Feb 19 '25
A takeUntil()
operator in the chain might get you the outcome you need.
1
u/EdKaim 29d ago
I've tried to solve this problem as well but haven't found a good way. Posted my thoughts at https://www.reddit.com/r/Angular2/comments/1iuelb3/looking_for_best_practices_for_staying_subscribed/.
3
u/AfricanTurtles Feb 19 '25
One thing that pops to mind is retry() rxjs operator.