Merging a new rxjs observable into existing subscribed observable

Question

I have an rxjs observable which is already subscribed to:

const o = new rxjs.Subject();
o.subscribe( (v) => console.log("value: " + v);

Due to some unrelated event, e.g. a user clicking on a button, I need to go and fetch data from the server, and put that data into the existing observable o. Using axios, I have something like this:

const aPromise = axios.get("someurl...")

I then do the following:

const secondObservable = rxjs.of(aPromise)
                             .subscribe( (response) => o.next(response.data) )

My question is whether there is a better way to put the response data into the existing already subscribed to observable, perhaps using operators?

Perhaps axios.get(...).then(response => o.next(response.data)).catch(error => o.error(error); — Richard Matsen

kos kos · Accepted Answer · 2019-04-15T21:03:45

If you know that you need data from your original stream and might get updates from a button click — then we'll need to create a stream of button clicks, e.g.

const btnClick$ = new rxjs.Subject();

onClick(){
  this.btnClick$.next(void 0);
}

Then we'll turn this btnClick$ stream into a http-get request stream, using switchMap:

const request$ = btnClick$.pipe(
  // each click will be turned into a http-get
  switchMap(() => rxjs.of(axios.get("someurl...")))
)

And we're ready to consume results from both streams, original and a stream of possible data updates from server on button click:

// we'll take results both from original stream
// and our new possible source of data
merge(original$, request$).subscribe(data => {
  console.log(data);
})

Heres a little app using mocks:

const { fromEvent, of, merge } = rxjs;
const { map, delay, switchMap } = rxjs.operators;

// create a stream of btn clicks
const theButton = document.getElementById('the-btn');
const btnClicks$ = fromEvent(theButton, 'click');

console.log('Processing...');

// clicks turned into http-gets
const request$ = btnClicks$.pipe(
  switchMap(() => makeRequest())
);

// original source of data
const original$ = of('Please, click the button!').pipe(delay(500));

merge(original$, request$)
  .subscribe(data => {
    console.log(data);
  });

// mock for requests
function makeRequest(){
  return of(Date.now()).pipe(delay(500))
}

<script src="https://unpkg.com/[email protected]/bundles/rxjs.umd.min.js"></script>

<button
  id="the-btn"
  >Click me!</button>

Hope this helps

Merging a new rxjs observable into existing subscribed observable

2 Answers