How to use rxjs buffer with takeWhile

Question

I am working on webrtc. The application sends icecandidates to backend firestore server. The problem is the call to signaling server is made multiple times as onicecandidate is triggered multiple time. I want collect all the icecandidates and make a single call to signaling server. The idea is to buffer all the events untill iceGathering is finished. This below attempt does not work

this.pc = new RTCPeerConnection(iceServers);
const source: Observable<any> =  fromEvent(this.pc, 'icecandidate');
const takeWhile$ = source
        .pipe(
            takeWhile(val=> val.currentTarget.iceGatheringState === 'gathering'
    ))
const buff = source.pipe(buffer(takeWhile$));
    buff.subscribe(() => {
        // this.pc.onicecandidate = onicecandidateCallback;
    })

So do you want to wait until gathering is finished then send the last value? Also where does onicecandidateCallback this come from? Is the problem basically "wait until status is not gathering" then assign this callback? — paulpdaniels
– paulpdaniels, Commented Jun 23, 2020 at 9:18

bharat.goli · Accepted Answer · 2020-06-22 04:19:23Z

Method 1:

You are almost there.

The takeWhile$ takes values and emits them while condition is met. So in buff, whenever takeWhile$ emits a value, buff emits a buffer of icecandidate events.

So you only need to emit one value in takeWhile$.

So what you need is takeLast() operator to only emit the last value.

When you put takeLast(1) in takeWhile$, it only emits last value and in buff, last emitted value leads to creation of buffer of icecandidate events.

this.pc = new RTCPeerConnection(iceServers);

const source: Observable<any> = fromEvent(this.pc, "icecandidate");

const takeWhile$ = source.pipe(
  takeWhile(val => val.currentTarget.iceGatheringState === "gathering"),
  takeLast(1)
);

const buff = source.pipe(buffer(takeWhile$));

buff.subscribe((bufferValues) => {

   // bufferValues has a buffer of icecandidate events

  // this.pc.onicecandidate = onicecandidateCallback;
});

You'll have access to buffer of icecandidate events in the subscription as bufferValues in above code.

Method 2:

You can also use reduce operator to achieve same scenario

this.pc = new RTCPeerConnection(iceServers);

const source: Observable<any> = fromEvent(this.pc, "icecandidate");

const takeWhile$ = source.pipe(
  takeWhile(val => val.currentTarget.iceGatheringState === "gathering"),
  reduce((acc, val) => [...acc,val], [])
);

takeWhile$.subscribe((bufferValues) => {

  // bufferValues has a buffer of icecandidate events

 // this.pc.onicecandidate = onicecandidateCallback;
})

Collectives™ on Stack Overflow

How to use rxjs buffer with takeWhile

1 Answer 1

Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

Comments

Your Answer

Sign up or log in

Post as a guest

Related