In my Angular app I would like to get SSE events from a server, and then do something with the results. For this, I found a solution where I wrap the SSE EventSource into an Observable. The code is the following:

import { Injectable, NgZone } from '@angular/core';
import { Observable } from 'rxjs';

@Injectable({
  providedIn: 'root',
})
export class SseServiceService {
  constructor(private _zone: NgZone) {}

  /**
   * Creates an event source
   */
  getEventSource(url: string): EventSource {
    return new EventSource(url);
  }

  /**
   * Returns an event source stream from the url
   *
   * @param url url of the event source
   */
  getServerSentEvent(url: string) {
    return new Observable((observer) => {
      const eventSource = this.getEventSource(url);

      eventSource.onmessage = (event) => {
        this._zone.run(() => {
          observer.next(event);
        });
      };

      eventSource.onerror = (error) => {
        this._zone.run(() => {
          observer.error(error);
        });
      };
    });
  }
}

The question is:

Should't I call eventSource.close() when the observable is getting destoryed?

Is there a way to assign a destructor, to observables made with new Observable()?

CodePudding user response：

You can optionally return a teardown function form the "subscribe function" passed to the constructor:

return new Observable((observer) => {
  const eventSource = this.getEventSource(url);
  ...
  return () => eventSource.close();
})

There're also operators such as finalize() or tap() (in RxJS 7 ) that let you call a function when the chain is being disposed.

CodePudding user response：

Hopefully martin's answer is what you need. But in the worst case, if for some reason you can't do it that way, it's possible in modern environments to get a callback in most cases when an object is going to be removed from memory because it's no longer referenced by anything and garbage collection is being done using a FinalizationRegistry. But note the warning in the docs:

Avoid where possible

Correct use of FinalizationRegistry takes careful thought, and it's best avoided if possible. It's also important to avoid relying on any specific behaviors not guaranteed by the specification. When, how, and whether garbage collection occurs is down to the implementation of any given JavaScript engine. Any behavior you observe in one engine may be different in another engine, in another version of the same engine, or even in a slightly different situation with the same version of the same engine. Garbage collection is a hard problem that JavaScript engine implementers are constantly refining and improving their solutions to.

(Disclosure: I wrote those docs, in consultation with the TC39 members behind the proposal that added it to JavaScript.)

That said, in the normal course of things, in a modern environment, you should get the callback if garbage collection happens. Here's how you'd do that (see comments):

export class SseServiceService {
    /**
     * A registry for `EventSource` cleanup, see `getServerSentEvent`.
     */
    private sseRegistry = new FinalizationRegistry((eventSource) => {
        eventSource.close();
    });

    constructor(private _zone: NgZone) { }

    /**
     * Creates an event source
     */
    getEventSource(url: string): EventSource {
        return new EventSource(url);
    }

    /**
     * Returns an event source stream from the url
     *
     * @param url url of the event source
     */
    getServerSentEvent(url: string) {
        const eventSource = this.getEventSource(url);
        const observable = new Observable((observer) => {
            eventSource.onmessage = (event) => {
                this._zone.run(() => {
                    observer.next(event);
                });
            };

            eventSource.onerror = (error) => {
                this._zone.run(() => {
                    observer.error(error);
                });
            };
        });
        // Register the observable (held weakly) and the
        // event source that will be passed to the cleanup
        // callback
        this.sseRegistry.register(observable, eventSource);
    }
}

And here's a live example of a cleanup callback getting called (in modern environments):

const registry = new FinalizationRegistry((heldValue) => {
    console.log(`Cleanup callback for ${heldValue} was called`);
});

console.log("Adding objects...");
let o1 = Array.from({length: 10_000}, () => ({}));
registry.register(o1, "o1");
let o2 = Array.from({length: 10_000}, () => ({}));
registry.register(o2, "o2");
let o3 = Array.from({length: 10_000}, () => ({}));
registry.register(o3, "o3");

setTimeout(() => {
    console.log("Releasing o1 and o3...");
    o1 = null;
    o3 = null;
}, 800);
setTimeout(() => {
    console.log("Encouraging garbage collection by allocating a large object");
    Array.from({length: 1_000_000}, () => ({}));
}, 1600);

For me, using a Chromium browser with the V8 engine, I don't see the cleanup callbacks at all (well, I waited about a minute) unless we "encourage" garbage collection by allocating a large object. In contrast, with SpiderMonkey in Firefox, I see the cleanup happen after a several second delay even without encouraging garbage collection. This underscores how the cleanup callback may get called much later than when the object is no longer referenced (or not at all), and how it varies by implementation.

So again, only do this if you can't find any other way. Hopefully you can use martin's answer or something similar instead.