如何从 onmessage 回调中创建一个可观察对象?

How do I create an observable from the onmessage callback?

我已经相当习惯在 .NET 中使用 RX 和 Java,我希望能够执行以下操作:

Rx.Observable.fromCallback(websocket.onmessage)
    .map(...)
    .subscribe(...);

然而,控制台有以下内容:

Uncaught TypeError: Rx.Observable.fromCallback(websocket.onmessage).map is not a function

这似乎表明 fromCallback 没有返回 Observable。

我在这里做错了什么?我是否误解了 fromCallback 在做什么,我需要使用 Subject?我不能将一些任意处理程序包装在可观察对象中吗?

你实际上是在寻找fromEvent or fromEventPattern:

Rx.Observable.fromEvent(websocket, 'message').map(/*...*/).subscribe();

Rx.Observable.fromEventPattern(
  function add(h) { websocket.addEventListener(h); }, 
  function remove(h) { websocket.removeEventListener(h); })
 .map(/*...*/)
 .subscribe();

第一个将尝试使用一些标准方法来订阅事件发射器,WebSocket 就是这样。但是,如果失败,您可以使用 fromEventPattern 来指定如何在对象中添加或删除处理程序。

另外请注意,Java脚本不会像 C# 和 Java 那样传递对您正在使用的对象实例的隐式引用,因此您的代码 fromCallback(websocket.onmessage) 是不传递 websocket,它传递函数原型中对方法的引用。 this 将在执行时确定。

Rx.Observable.fromCallback 适用于最后一个参数是回调函数的函数,回调函数是异步 Java 脚本代码的标准模式。此外,fromCallback 方法不 return 一个 Observable 它 return 是一个函数,当被调用时 return 是一个 Observable

function methodWithCallback(arg0, arg1, cb) {
  setTimeout(function() {
    cb(arg0 + arg1);
  }, 2000);
}

var newMethod = Rx.Observable.fromCallback(methodWithCallback);

//[After 2 seconds] 3
newMethod(1, 2).subscribe(console.log.bind(console));