Redux Observable：如何从回调中 return 一个动作？

Question

我正在使用具有非常具体 API 的 WebRTC 库。 peerConnection.setRemoteDescription 方法的第二个参数应该是它完成设置远程描述时的回调：

这是我的 WebRTC 包装函数之一 class:

export function setRemoteSdp(peerConnection, sdp, callback) {
  if (!sdp) return;
  return peerConnection.setRemoteDescription(
    new RTCSessionDescription(sdp),
    callback, // <-------------
  );
}

这是我想要做的草图：

function receivedSdp(action$, store) {
  return action$.ofType(VideoStream.RECEIVED_SDP)
    .mergeMap(action => {
      const {peerConnection} = store.getState().videoStreams;
      const {sdp} = action.payload;

      return WebRTC.setRemoteSdp(peerConnection, sdp, () => {
        return myReducer.myAction(); // <------ return action as the callback
      })
    })
};

这不起作用，因为我没有返回 Observable。有办法吗？

P.S。这是 WebRTC API：https://github.com/oney/react-native-webrtc/blob/master/RTCPeerConnection.js#L176

Answer 1

所以问题是 setRemoteSdp 不是 return 一个 Observable，而 myReducer.myAction() 是一个 Observable，这就是你想要合并的 Observable？

您可以使用 Observable.create 并包装 WebRTC.setRemoteSdp 调用：

.mergeMap(action => {
  return Observable.create(observer => {
    WebRTC.setRemoteSdp(peerConnection, sdp, () => {
      observer.next(myReducer.myAction());
      observer.complete();
    })
  });
}
.mergeAll()

Observable.create return 是一个从 myReducer.myAction() 发出另一个 Observable 的 Observable。现在我实际上有了所谓的高阶，我想使用 mergeAll() 展平（concatAll 也可以）。

Answer 2

马丁关于使用 Observable.create 或 new Observable 的回答是正确的——同样的事情（除了我不清楚为什么你需要 mergeAll() 因为 mergeMap 会扁平化？）

作为奖励，您还可以为此使用 Observable.bindCallback。

// bindCallback is a factory factory, it creates a function that
// when called with any arguments will return an Observable that
// wraps setRemoteSdp, handling the callback portion for you.
// I'm using setRemoteSdp.bind(WebRTC) because I don't know
// if setRemoteSdp requires its calling context to be WebRTC
// so it's "just in case". It might not be needed.
const setRemoteSdpObservable = Observable.bindCallback(WebRTC.setRemoteSdp.bind(WebRTC));

setRemoteSdpObservable(peerConnection, sdp)
  .subscribe(d => console.log(d));

你的史诗中的用法是这样的

// observables are lazy, so defining this outside of our epic
// is totally cool--it only sets up the factory
const setRemoteSdpObservable = Observable.bindCallback(WebRTC.setRemoteSdp.bind(WebRTC));

function receivedSdp(action$, store) {
  return action$.ofType(VideoStream.RECEIVED_SDP)
    .mergeMap(action => {
      const {peerConnection} = store.getState().videoStreams;
      const {sdp} = action.payload;

      return setRemoteSdpObservable(peerConnection)
        .map(result => myReducer.myAction());
    })
};

您可以使用它为所有 WebRTC api 创建 Observable 包装器。

Redux Observable：如何从回调中 return 一个动作？

Redux Observable: How to return an action from a callback?

rxjs

redux-observable