javascript - Node JS SSE: Το σφάλμα του παρατηρητή RxJs αγνοείται ενώ μοιράζεται μια συνδρομή

Υπάρχει ένα τέλος SSE που μοιράζεται μια συνδρομή εάν ο καταναλωτής με το ίδιο κλειδί είναι ήδη εγγεγραμμένος. Εάν υπάρχει ενεργή συνδρομή, τα δεδομένα αντλούνται από άλλο πελάτη.

Το πρόβλημα είναι ότι η εξωτερική συνδρομή φαίνεται να μην ανιχνεύει ποτέ το σφάλμα και να το αναθέτει στον δρομολογητή ώστε να κλείσει η σύνδεση με τον πελάτη: η παρακολούθηση σταματά, αλλά η σύνδεση παραμένει ενεργή.

Νομίζω το πρόβλημα είναι πώς ξεκινάω τη συνδρομή που πρόκειται να κοινοποιηθεί... αλλά αυτή τη στιγμή δεν μπορώ να σκεφτώ έναν άλλο τρόπο να το λύσω.

Δρομολογητής (SSE)/εξωτερική συνδρομή:

...

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);
});

...

ΥπηρεσίαΣυνδρομής

...

@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;
}
...

Υπηρεσία Δημοσκόπησης

...

@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}"`);
    }
}
...

Απάντηση #1

throwError δεν προκαλεί πραγματικά σφάλμα, αλλά δημιουργεί ένα observable που εκπέμπει ένα σφάλμα.

Από τα έγγραφα:

[throwError] Δημιουργεί ένα observable που θα δημιουργήσει ένα στιγμιότυπο σφάλματος και θα το στείλει στον καταναλωτή ως σφάλμα αμέσως μετά τη συνδρομή.

Γι' αυτό η χρήση του μέσα σε subscribe δεν λειτουργεί όπως αναμένεται. Απλώς θα πρέπει να κάνετε throw:

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

Φαίνεται ότι έχετε κάποια περιττή πολυπλοκότητα στον τρόπο που καλείτε το fetchData() προκειμένου να εγγραφείτε και να στείλετε το αποτέλεσμα σε ένα BehaviorSubject. Δεν γνωρίζω όλες τις απαιτήσεις σας, αλλά φαίνεται ότι ίσως δεν χρειάζεστε καθόλου το BehaviorSubject.

Αντί να εγγράφεστε στο fetchData(), θα μπορούσατε απλώς να επιστρέψετε το παρατηρήσιμο και να το προσθέσετε στον πίνακα pool$, ή ίσως ακόμη και να καταργήσετε τελείως το fetchData():

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}"`);
    }
}

Σημειώσεις :

• Με την παραπάνω απλοποίηση, ίσως να μην χρειάζεσαι πλέον το εξωτερικό try/catch
• αυτό δεν είναι μια πλήρης λύση και μπορεί να απαιτεί κάποιες τροποποιήσεις σε άλλα σημεία του κώδικά σας. Ήθελα απλώς να επισημάνω αυτό που φαίνεται σαν περιττή πολυπλοκότητα.