Из

В отдельном пакете RxJavaAsyncUtil, который не включен по умолчанию в RxGroovy, также существует функция runAsync. Передайте runAsync Action и Scheduler, и она вернёт StoppableObservable, которая использует указанный Action для генерации излучаемых элементов.

Action принимает Observer и Subscription. Она использует Subscription для проверки условия isUnsubscribed, после чего прекратит излучение элементов. Вы также можете вручную остановить StoppableObservable в любое время, вызвав метод unsubscribe (который также отменит подписку Subscription, связанную с StoppableObservable).

Поскольку runAsync немедленно вызывает Action и начинает излучать элементы, возможно, что некоторые элементы могут быть потеряны в промежутке между установлением StoppableObservable с помощью этого метода и готовностью вашего Observer к получению элементов. Если это проблема, вы можете использовать вариант runAsync, который также принимает Subject, и передать ReplaySubject для извлечения пропущенных элементов.

В отдельном пакете RxJavaAsyncUtil, который не включен по умолчанию в RxJava, также существует функция runAsync. Передайте runAsync Action и Scheduler, и она вернёт StoppableObservable, которая использует указанный Action для генерации излучаемых элементов.

Action принимает Observer и Subscription. Она использует Subscription для проверки условия isUnsubscribed, после чего прекратит излучение элементов. Вы также можете вручную остановить StoppableObservable в любое время, вызвав метод unsubscribe (который также отменит подписку Subscription, связанную с StoppableObservable).

Поскольку runAsync немедленно вызывает Action и начинает излучать элементы, возможно, что некоторые элементы могут быть потеряны в промежутке между установлением StoppableObservable с помощью этого метода и готовностью вашего Observer к получению элементов. Если это проблема, вы можете использовать вариант runAsync, который также принимает Subject, и передать ReplaySubject для извлечения пропущенных элементов.

Существует несколько специализированных вариантов оператора From в RxJS:

Также существует оператор fromPromise, который преобразует Promise в Observable, преобразуя вызовы `then` в испускания `next`, и вызовы `catch` в испускания `error`.

Оператор 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