From

待定

待定

在单独的 RxJavaAsyncUtil 包中（默认情况下不包含在 RxGroovy 中），还有一个 runAsync 函数。将 runAsync 传递给 Action 和一个 Scheduler，它将返回一个 StoppableObservable，该 StoppableObservable 使用指定的 Action 来生成它发射的项目。

Action 接受一个 Observer 和一个 Subscription。它使用 Subscription 来检查 isUnsubscribed 条件，如果该条件成立，它将停止发射项目。您也可以随时通过调用其 unsubscribe 方法来手动停止 StoppableObservable（这也会取消您与 StoppableObservable 关联的 Subscription 的订阅）。

因为 runAsync 会立即调用 Action 并开始发射项目，所以有可能在您使用此方法建立 StoppableObservable 和您的 Observer 准备接收项目之间的时间间隔内会丢失一些项目。如果这是个问题，您可以使用 runAsync 的变体，该变体还接受一个 Subject，并将 ReplaySubject 传递给它，您可以使用该 ReplaySubject 来检索其他情况下丢失的项目。

在单独的 RxJavaAsyncUtil 包中（默认情况下不包含在 RxJava 中），还有一个 runAsync 函数。将 runAsync 传递给 Action 和一个 Scheduler，它将返回一个 StoppableObservable，该 StoppableObservable 使用指定的 Action 来生成它发射的项目。

Action 接受一个 Observer 和一个 Subscription。它使用 Subscription 来检查 isUnsubscribed 条件，如果该条件成立，它将停止发射项目。您也可以随时通过调用其 unsubscribe 方法来手动停止 StoppableObservable（这也会取消您与 StoppableObservable 关联的 Subscription 的订阅）。

因为 runAsync 会立即调用 Action 并开始发射项目，所以有可能在您使用此方法建立 StoppableObservable 和您的 Observer 准备接收项目之间的时间间隔内会丢失一些项目。如果这是个问题，您可以使用 runAsync 的变体，该变体还接受一个 Subject，并将 ReplaySubject 传递给它，您可以使用该 ReplaySubject 来检索其他情况下丢失的项目。

待定

RxJS 中有几个专门的 From 变体

还有一个 fromPromise 操作符，它将 Promise 转换为 Observable，将它的 resolve 调用转换为 onNext 通知，并将它的 reject 调用转换为 onError 通知。

fromPromise 存在于以下发行版中

• rx.async.js（需要 rx.binding.js 和 rx.js 或 rx.compat.js）
• rx.async.compat.js（需要 rx.binding.js 和 rx.js 或 rx.compat.js）
• rx.lite.js
• rx.lite.compat.js

示例代码

var promise = new RSVP.Promise(function (resolve, reject) {
   resolve(42);
});

var source = Rx.Observable.fromPromise(promise);

var subscription = source.subscribe(
    function (x) { console.log('Next: ' + x); },
    function (e) { console.log('Error: ' + e); },
    function ( ) { console.log('Completed'); });

Next: 42:
Completed

var promise = new RSVP.Promise(function (resolve, reject) {
   reject(new Error('reason'));
});

var source = Rx.Observable.fromPromise(promise);

var subscription = source.subscribe(
    function (x) { console.log('Next: ' + x); },
    function (e) { console.log('Error: ' + e); },
    function ( ) { console.log('Completed'); });

Error: Error: reject

还有一个 ofArrayChanges 操作符，它使用 Array.observe 方法监控数组，并返回一个发射数组中发生的任何更改的 Observable。此操作符仅存在于 rx.all.js 发行版中。

示例代码

var arr = [1,2,3];
var source = Rx.Observable.ofArrayChanges(arr);

var subscription = source.subscribe(
    function (x) { console.log('Next: ' + x); },
    function (e) { console.log('Error: ' + e); },
    function ( ) { console.log('Completed'); });

arr.push(4)

Next: {type: "splice", object: Array[4], index: 3, removed: Array[0], addedCount: 1}

类似的操作符是 ofObjectChanges。它返回一个发射对特定对象进行的任何更改的 Observable，如 Object.observe 方法报告的那样。它也仅存在于 rx.all.js 发行版中。

示例代码

var obj = {x: 1};
var source = Rx.Observable.ofObjectChanges(obj);

var subscription = source.subscribe(
    function (x) { console.log('Next: ' + x); },
    function (e) { console.log('Error: ' + e); },
    function ( ) { console.log('Completed'); });

obj.x = 42;

Next: {type: "update", object: Object, name: "x", oldValue: 1}

还有一个 pairs 操作符。此操作符接受一个对象，并返回一个发射该对象的属性作为键值对的 Observable。

示例代码

var obj = {
  foo: 42,
  bar: 56,
  baz: 78
};

var source = Rx.Observable.pairs(obj);

var subscription = source.subscribe(
    function (x) { console.log('Next: ' + x); },
    function (e) { console.log('Error: ' + e); },
    function ( ) { console.log('Completed'); });

Next: ['foo', 42]
Next: ['bar', 56]
Next: ['baz', 78]
Completed

pairs 存在于以下发行版中

• rx.js
• rx.all.js
• rx.all.compat.js
• rx.compat.js
• rx.lite.js
• rx.lite.compat.js

待定

待定

待定

待定

待定

在 Swift 中，这是使用 Observable.from 类方法实现的。

数组中的每个元素都作为发射产生。此方法与 Observable.just 之间的区别在于后者将整个数组作为一个发射发射。

示例代码

let numbers = [1,2,3,4,5]

let source = Observable.from(numbers)

source.subscribe {
    print($0)
}

next(1)
next(2)
next(3)
next(4)
next(5)
completed