如何在takeWhile中使用rxjs缓冲区

debugcn 发表于 Dev

节点马里奥

我正在研究webrtc。该应用程序将icecandidates发送到后端Firestore服务器。问题在于，多次触发onicecandidate导致多次调用信令服务器。我想收集所有的候选人，并打一个电话到信令服务器。这个想法是缓冲所有事件，直到iceGathering完成。这下面的尝试不起作用

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;
    })

巴拉特1226

方法1：

你快到了。

该takeWhile$取值，发出他们，而条件得到满足。因此，在中buff，无论何时takeWhile$发出值，都会发出事件buff缓冲区icecandidate。

因此，您只需在中发出一个值takeWhile$。

因此，您需要的是takeLast()运算符仅发出最后一个值。

当您放入takeLast(1)in时takeWhile$，它仅发出最后一个值，而在in中buff，最后发出的值导致创建icecandidate事件缓冲区。

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;
});

icecandidate如bufferValues上面的代码，您将可以访问订阅中的事件缓冲区。

方法2：

您还可以使用reduce运算符来实现相同的方案

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;
})