CODEWITHSUNDEEP
rxjs
doe
doe

Reputation: 455

Combining multiple streams/observables to use mapping operators

I have 2 servers with separate streams/observables, how can I make both use the same chain so I can control them with switchMap, flatMap etc operators?

if (serverNumber == 1) {
  this.streamService.getStream('test 1')
    .pipe(
      map(image => console.log(image))
    ).subscribe();
}
else  {
  this.streamService.getFastStream()
    .pipe(
      map(image => console.log(image))
    ).subscribe();
}

For example if I fetch from server and getStream takes a while and is still in progress after getFastStream completes I want to block overwrite by the first getStream with switchMap. How can I do this? I'm new to rxjs so what I'm trying might not be how its supposed to be used.

The observable

  images$: Observable<string> = new Observable<string>();

Then do I have to push the stream onto the images$ observable with one of the operators?

this.images$.pipe(
  switchMap(this.streamService.getStream(1))
)

Template:

<div class="section">
    Click to send a request ...
</div>
<div class="section">
    <button (click)="click(1)">Server 1</button>
</div>
<div class="section">
    <button (click)="click(5)">Server 2 (Longer)</button>
</div>

Upvotes: 1

Views: 189

Answers (2)

benshabatnoam
benshabatnoam

Reputation: 7640

First use takeUntil operator on the fast observable. Second use merge to take fast observable results until slow observable results start to come.

import { merge } from 'rxjs';
import { takeUntil } from 'rxjs/operators';

const slow$ = this.streamService.getStream('test 1').pipe(
  map(image => console.log(image))
);

const fast$ = this.streamService.getFastStream().pipe(
  map(image => console.log(image)),
  takeUntil(slow$)
);

merge(
  fast$,
  slow$
).subscribe()

Here is a StackBlitz DEMO to demonstrate this.

Upvotes: 1

Pankaj
Pankaj

Reputation: 568

RXJS race observable creator might help.

var server1$ = this.streamService.getStream('test 1');
var fastServer$ = this.streamService.getFastStream();

var fastest$ = race(server1$, fastServer$)
              .pipe(map(image=> console.log(image)));
fastest$.subscribe();

Upvotes: 0

Related Questions