coffee> rext = require 'rx'
coffee> arr = [1..5]
[ 1, 2, 3, 4, 5 ]
coffee> obs = rext.Observable.fromArray(arr)
{ _subscribe: [Function] }
coffee> obs.subscribe( (x) -> console.log("added value: " + x))
added value: 1
added value: 2
added value: 3
added value: 4
added value: 5
{ isStopped: true,
observer:
{ isStopped: true,
_onNext: [Function],
_onError: [Function: defaultError],
_onCompleted: [Function: noop] },
m: { isDisposed: true, current: null } }
coffee> arr.push(12) # expecting "added value: 12"
6 # instead got new length of array
coffee>
Похоже, что функция subscribe будет срабатывать только один раз, когда она будет создана. Похоже, что это немного неправильно, поскольку я действительно просто для каждого массива вместо наблюдения за изменениями на нем. Этот код почти точно такой же, как и на вики. Поэтому либо я делаю это неправильно, либо subscribe не работает, как я ожидаю.
Observable.fromArray создает Observable, который немедленно запускает события для каждого элемента массива, когда вы добавляете подписчика. Таким образом, он не будет "наблюдать" за изменениями в этом массиве.
Если вам нужна "pushable collection", класс Bus в Bacon.js может быть тем, что вы ищете. Для RxJs есть мой маленький MessageQueue класс, который имеет аналогичную функциональность.
В RxJS то, что вы ищете, называется Subject.
Вы можете вставлять в него данные и передавать их оттуда.
Пример:
var array = [];
var arraySubject = new Rx.Subject();
var pushToArray = function (item) {
array.push(item);
arraySubject.next(item);
}
// Subscribe to the subject to react to changes
arraySubject.subscribe((item) => console.log(item));
Я нашел Rx.Observable.ofObjectChanges(obj) для работы так, как я ожидал.
На странице документации:
Создает наблюдаемую последовательность из изменений объекта с помощью Object.observe.
Надеюсь, что это поможет.