Angular / RxJS Multicasting Observable of Observable

Question

I have an HTTP request that I want to share the results of to multiple components. The HTTP request, of course, returns an Observable. I want multiple components to be able to subscribe to this without triggering additional HTTP requests.

I accomplished this using a Subject in a component that made the HTTP request on demand and has another method to subscribe to the subject. While this works - it seems like overkill and there certainly is a better way of doing this.

subject service

@Injectable()
export class EventService {
    subject: BehaviorSubject<any> = new BehaviorSubject<any>(Observable.create());

    constructor(private api: Api) {}

    fetch = () => {
        this.subject.next(
            this.api.getUsers().pipe(share())
        );
    };

    listen = (): Observable<any> => {
        return this.subject.asObservable();
    };
}

and one subscriber

@Injectable
export class EventListenerA {
    constructor(private eventService: EventService){
        eventService.fetch(); // make initial call
        eventService.listen().subscribe(o => {
             o.subscribe(httpResponse => {
                 //do something with response
             })
        });
    }
}

and second subscriber

@Injectable
export class EventListenerB {
    constructor(private eventService: EventService){
        eventService.listen().subscribe(o => {
             o.subscribe(httpResponse => {
                 //do something else with response
             })
        });
    }
}

When I remove share() from the pipe chain, multiple network requests are made. Is there a more elegant / correct way of passing an observable to the next of a Subject ? Or another pattern altogether

Reactgular Reactgular · Accepted Answer · 2019-01-15T01:05:25

You should use the ReplaySubject() without giving it an initial value. This way when a component subscribes to listen() it will wait until there is data available.

There is no need to share the HTTP observable with other components. Just subscribe to the HTTP request and then send the value to the ReplaySubject(). Any components listening will receive the data.

@Injectable()
export class EventService {
     private _events: ReplaySubject<any> = new ReplaySubject(1);

     public constructor(private _api: ApiService) {}

     public fetch() {
         this.api.getUsers().subscribe(value => this._events.next(value));
     }

     public listen() {
         return this._events.asObservable();
     }
 }

 @Injectable
 export class EventListenerA {
     constructor(private eventService: EventService){
          eventService.fetch();
          eventService.listen().subscribe(o => {
              // prints user data
              console.log(o); 
          });
     }
}

Everytime someone calls eventService.fetch() it will trigger a HTTP request. If you one want to perform this request once, then maybe call fetch() from the service constructor or another spot in the application that happens only once (i.e. a module constructor).

Angular / RxJS Multicasting Observable of Observable

1 Answers