Node JS SSE: RxJs observable error gets ignored while sharing a subscription

user_3693726 0
Viewed 113

There is an SSE endpoint that shares a subscription if the consumer with the same key is already subscribed. If there is an active subscription the data is being polled from another client.

The problem is that the outer subscription never seems to catch the error and delegate it to the router in order to close the connection with the client: polling stops, but connection stays active.

I think the issue is how I start the subscription that is to be shared... but I can't think of a way to resolve this in another way currently.

Router (SSE) / outer subscription:

...

const clientId = Date.now();
const newClient = {
    id: clientId,
    res,
};
clients.push(newClient);

const sub = subscriptionService.listenToChanges(req.context, categoryIds).subscribe({
    next: (data) => {
        if (JSON.stringify(data) !== '{}') {
            newClient.res.write(`data: ${JSON.stringify(data)}\n\n`);
        } else {
            newClient.res.write(': poke...\n\n');
        }
    },
    error: () => {
        // we never get here...
        next(new InternalError());
        clients = clients.filter((c) => c.id !== clientId);
        res.end();
    },
    complete: () => {
        res.end();
        clients = clients.filter((c) => c.id !== clientId);
    },
});

req.on('close', () => {
    subscriptionService.stopListening(req.context);
    sub.unsubscribe();
    clients = clients.filter((c) => c.id !== clientId);
});

...

SubscriptionService

...

@trace()
public listenToChanges(ctx: Context, ids: string[]): Observable<{ [key: string]: Data }> {
    const key = ctx.user?.email || ClientTypeKey.Anon;
    try {
        if (this.pool$[key]) {
            return this.pool$[key];
        }


        this.poolSource[key] = new BehaviorSubject<{ [p: string]: Data }>({});
        this.pool$[key] = this.poolSource[key].asObservable();

        this.fetchData(ctx, ids);

        return this.pool$[key].pipe(
            catchError((e) => throwError(e)), // we never get here...
        );
    } catch (e) {
        throw new Error(`Subscription Service Listen returned an error: "${e}"`);
    }
}

...

private fetchData(ctx: Context, ids: string[]): void {
    const key = ctx.user?.email || ClientTypeKey.Anon;

    const sub = this.service.getData(ctx, ids)
        .pipe(
            catchError((e) => throwError(e)),
        ).subscribe(
            (r) => this.poolSource[key].next(r),
            (e) => throwError(e), // last time the error is caught
        );
    this.subscriptions[key] = sub;
}
...

Polling Service

...

@trace()
public getData(ctx: Context, ids: string[]): Observable<{[key: string]: Data}> {
    try {
        const key = ctx.user?.email || ClientTypeKey.Anon;
        const pollingInterval = config.get('services.pollingInterval') || 10000;

        return interval(pollingInterval).pipe(
            startWith(0),
            switchMap(() => this.getConfig(ctx, !!this.cachedData[key])),
            map((r) => this.getUpdatedData(ctx, r.data, ids)),
            catchError((e) => throwError(e)),
        );
    } catch (e) {
        throw new Error(`Get Data returned an error: "${e}"`);
    }
}
...
rxjs typescript node.js javascript
1 Answers

throwError doesn't actually throw an error, but rather creates an observable that emits an error.

From the docs:

[throwError] Creates an observable that will create an error instance and push it to the consumer as an error immediately upon subscription.

This is why using it inside subscribe does not work as intended. You should simply throw:

.subscribe(
  (r) => this.poolSource[key].next(r),
  (e) => throw new Error(e)
);

It seems like you have some unnecessary complexity in the way you are calling fetchData() in order to subscribe and push the result into a BehaviorSubject. I don't know all your requirements, but it seems like maybe you don't need the BehaviorSubject at all.

Instead of subscribing in fetchData(), you could simply return the observable and add that into your pool$ array, or maybe even get rid of fetchData() altogether:

public listenToChanges(ctx: Context, ids: string[]): Observable<{ [key: string]: Data }> {
    const key = ctx.user?.email || ClientTypeKey.Anon;
    try {
        if (!this.pool$[key]) {
            this.pool$[key] = this.service.getData(ctx, ids).pipe(
                catchError((e) => throwError(e))
            );
        }

        return this.pool$[key];
    } catch (e) {
        throw new Error(`Subscription Service Listen returned an error: "${e}"`);
    }
}

Notes:

  • with the above simplification, maybe you no longer need the outer try/catch
  • this isn't a complete solution and may require some tweaks in other places of your code. I just wanted to point out, what seems like unnecessary complexity.
user_18583570
Related