[解決済み] RxJS：Observableを「手動で」更新するにはどうしたらいいですか？

質問

私の考えでは、これは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') しかし、それは実際には機能しないように思えたし、オブザーバーで呼び出されているので、正しいとは思えません。オブザーバーはデータストリームに反応するものであって、それを変更するものではありませんよね？

私がオブザーバーとオブザーバブルの関係を理解していないだけなのでしょうか？

どのように解決するのですか？

RXでは、ObserverとObservableは別個のエンティティです。オブザーバーは、オブザーバブルを購読します。Observableは、Observerのメソッドを呼び出すことで、そのObserverにアイテムを発行します。のスコープ外でオブザーバのメソッドを呼び出す必要がある場合、そのオブザーバは Observable.create() を使用することができます。これは、オブザーバと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'); 
}

被写体については、こちらで詳しく解説しています。

• https://github.com/ReactiveX/rxjs/blob/master/docs_app/content/guide/subject.md
• http://reactivex.io/documentation/subject.html