RxJS: Как мне "вручную" обновить Observable?

Я думаю, что я, должно быть, неправильно понимаю что-то фундаментальное, потому что, на мой взгляд, это должно быть самым основным случаем для наблюдаемого, но для моей жизни я не могу понять, как это сделать из документов.

В принципе, я хочу быть в состоянии сделать это:

// create a dummy observable, which I would update manually
var eventObservable = rx.Observable.create(function(observer){});
var observer = eventObservable.subscribe(
   function(x){
     console.log('next: ' + x);
   }
...
var my_function = function(){
  eventObservable.push('foo'); 
  //'push' adds an event to the datastream, the observer gets it and prints 
  // next: foo
}

Но я не смог найти метод, как push. Я использую это для обработчика кликов, и я знаю, что для этого у них есть Observable.fromEvent, но я пытаюсь использовать его с React, и я предпочел бы просто обновить поток данных в обратном вызове вместо использования совершенно другая система обработки событий. В общем, я хочу это:

$( "#target" ).click(function(e) {
  eventObservable.push(e.target.text()); 
});

Самое близкое, что я получил, - это использовать observer.onNext('foo'), но, похоже, это не сработало, и это вызвало наблюдателя, что кажется неправильным. Наблюдатель должен реагировать на поток данных, а не изменять его, верно?

Я просто не понимаю отношения наблюдатель/наблюдаемый?

Ответ 1

В RX Observer и Observable являются различными объектами. Наблюдатель подписывается на Observable. Observable передает объекты своим наблюдателям, вызывая методы наблюдателей. Если вам нужно вызывать методы наблюдателя вне области Observable.create() вы можете использовать Subject, который является прокси-сервером, который действует как наблюдатель и Observable одновременно.

Вы можете сделать так:

var eventStream = new Rx.Subject();

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

var my_function = function() {
  eventStream.next('foo'); 
}

Вы можете найти больше информации о предметах здесь:

Ответ 2

Я полагаю, что Observable.create() принимает не наблюдателя в качестве параметра обратного вызова, а излучателя. Так что, если вы хотите добавить новое значение в Observable, попробуйте это:

var emitter;
var observable = Rx.Observable.create(e => emitter = e);
var observer = {
  next: function(next) {
    console.log(next);
  },
  error: function(error) {
    console.log(error);
  },
  complete: function() {
    console.log("done");
  }
}
observable.subscribe(observer);
emitter.next('foo');
emitter.next('bar');
emitter.next('baz');
emitter.complete();

//console output
//"foo"
//"bar"
//"baz"
//"done"

Да Субъект делает это проще, предоставляя Observable и Observer в одном и том же объекте, но это не совсем одно и то же, поскольку Subject позволяет вам подписать несколько наблюдателей на одну и ту же наблюдаемую, когда наблюдаемая отправляет данные только последнему подписанному наблюдателю, поэтому используйте его сознательно, Вот JsBin, если вы хотите повозиться с этим.