[reactive-programming] 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')했지만 실제로는 작동하지 않는 것으로 보이며 관찰자에게 호출 된 것 같습니다. 관찰자는 데이터 스트림에 반응하는 것이지, 데이터 스트림을 변경하지 말아야합니다.

관찰자 / 관찰 가능한 관계를 이해하지 못합니까?

답변

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'); 
}

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"