Convert a collection of observables containing events to an observable array of events

Question

Convert a collection of observables containing events to an observable array of events

I am working on creating a function that merges all event streams generated from an array of data channels and emits any event received from them.

Below is the initial function:

  private dataChannels: BehaviorSubject<RTCDataChannel[]> = new BehaviorSubject([]);
  // The datachannels array will be populated at some point
  ...

  public on(): Observable<Event> {
    const eventStreams = this.dataChannels.value.map((channel) => fromEvent(channel, 'message'));
    return merge(...eventStreams);
  }

However, I realized that the issue with the current function is that it does not utilize new values emitted to the data channels BehaviorSubject.

As a result, I started developing the following function:

  const allEvents = this.dataChannels.pipe(map((channels) => channels.map((channel) => fromEvent(channel, 'message'))));

  return merge(...allEvents);

The problem arises when allEvents ends up being of type

Observable<Observable<Event>[]>

, which is not compatible with the merge function. How can I convert the observable to type Observable<Event>[]?

javascript angular typescript rxjs

Answer 1

Answer №1

To optimize your code, consider merging all the Observables received in a BehaviorSubject. There appears to be an issue with lingering subscriptions to channels even after they are removed from dataChannels. I suggest creating a new property named newSubscription$. Emitting this property will cause all subscriptions created using fromEvent to unsubscribe.

Give this solution a try:

    newSubscription$ = new Subject();
    this.dataChannels.pipe(
        tap(() => {
            this.newSubscription$.next();   // terminate existing subscriptions
            this.newSubscription$ = new Subject(); // and await for new ones
        }) 
        switchMap((allChannels) => {
        const fromEvenObs = allChannels.map(c => fromEvent(c, 'message').pipe(takeUntil(newSubscription$)));
        return merge(...fromEvenObs);
    })).subscribe((data) => { // data obtained from any event });

UPDATE:

Due to the usage of switchMap(), it is not necessary to have an additional observable to handle unsubscribing from events when new data is received by dataChannels. The following modification should suffice:

this.dataChannels.pipe(
    switchMap((allChannels) => {
        const fromEvenObs = allChannels.map(c => fromEvent(c, 'message'));
        return merge(...fromEvenObs);
})).subscribe((data) => { // data from events });

Answer 2

To optimize your code, consider merging all the Observables received in a BehaviorSubject. There appears to be an issue with lingering subscriptions to channels even after they are removed from dataChannels. I suggest creating a new property named newSubscription$. Emitting this property will cause all subscriptions created using fromEvent to unsubscribe.

Give this solution a try:

    newSubscription$ = new Subject();
    this.dataChannels.pipe(
        tap(() => {
            this.newSubscription$.next();   // terminate existing subscriptions
            this.newSubscription$ = new Subject(); // and await for new ones
        }) 
        switchMap((allChannels) => {
        const fromEvenObs = allChannels.map(c => fromEvent(c, 'message').pipe(takeUntil(newSubscription$)));
        return merge(...fromEvenObs);
    })).subscribe((data) => { // data obtained from any event });

UPDATE:

Due to the usage of switchMap(), it is not necessary to have an additional observable to handle unsubscribing from events when new data is received by dataChannels. The following modification should suffice:

this.dataChannels.pipe(
    switchMap((allChannels) => {
        const fromEvenObs = allChannels.map(c => fromEvent(c, 'message'));
        return merge(...fromEvenObs);
})).subscribe((data) => { // data from events });

Answer 3

Answer №2

To achieve this functionality, one can utilize the `mergeAll` operator according to the documentation's guidance that states `mergeAll subscribes to an Observable that emits Observables. Each time it observes one of these emitted inner Observables, it subscribes to that and delivers all the values from the inner Observable on the output Observable.`

Additionally, the `from` operator can be used to emit each observable within an array of observables.

const allEvents = this.dataChannels.pipe(
   map((channels) => from(channels.map((c) => fromEvent(c, 'message'))),
   mergeAll();
);

Answer 4

To achieve this functionality, one can utilize the `mergeAll` operator according to the documentation's guidance that states `mergeAll subscribes to an Observable that emits Observables. Each time it observes one of these emitted inner Observables, it subscribes to that and delivers all the values from the inner Observable on the output Observable.`

Additionally, the `from` operator can be used to emit each observable within an array of observables.

const allEvents = this.dataChannels.pipe(
   map((channels) => from(channels.map((c) => fromEvent(c, 'message'))),
   mergeAll();
);

Convert a collection of observables containing events to an observable array of events

Answer №1

Answer №2

Similar questions

Managed the double-click event to select Snap.svg text

Adjust the text color of ASP.Net Label based on the presence of a hyphen

Tips for customizing the Electron title bar and enabling drag functionality

Having trouble getting dynamic values to render properly in Ionic select? Find a solution in Ionic 4

Guide to utilizing JavaScript and JQuery for retrieving a PDF from a specific URL and subsequently uploading the file to another website

Error: Attempting to access the 'firstChild' property of a null object when trying to display a street view panorama with Vue.js

Javascript: Comparing Equality of two Objects/ Arrays

React JS, cumulative points total

Presenting quiz questions using javascript

AngularJS fails to recognize Iframe element during REST request

Create dynamic animations using AngularJS to transition between different states within an ng-repeat loop

Update the input component's typing area line color from its default shade

Activate a mouse click event based on user input in a Shiny

Troubleshooting a VueJS Problem: Updating $data in the mounted hook

Occasionally, altering the visibility of an element right before submitting a form in Safari may not be effective

Looking to attach a listener to an element within a modal once it has finished rendering?

Adding query parameters to an Angular route

Having trouble importing components from the module generated by Angular CLI library

nighttime surveillance struggled to locate the element based on its identifier

What is the proper way to set up state with localStorage in Nuxt.js when using Universal Rendering?