RxJS: Jak mogę "ręcznie" zaktualizować obserwowalne?

Question

Myślę, że muszę się nieporozumienie coś fundamentalnego, ponieważ w moim umyśle powinien to być najbardziej podstawowy przypadek dla obserwowalne, ale dla mojego życia nie mogę dowiedzieć się, jak to zrobić z dokumentów.

Zasadniczo, chcę być w stanie to zrobić:

// 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 
} 

Ale nie udało się znaleźć sposób jak push. Używam tego dla obsługi kliknięcia i wiem, że mają do tego Observable.fromEvent, ale próbuję użyć go z React i wolałbym po prostu zaktualizować strumień danych w wywołaniu zwrotnym, zamiast używać całkowicie inny system obsługi zdarzeń. Więc w zasadzie chcę to:

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

Najbliżej mam używał observer.onNext('foo'), ale to nie wydaje się, że rzeczywiście działa i nazywa na obserwatora, który nie wydaje się słuszne. Obserwator powinien być tym, który reaguje na strumień danych, a nie zmienia go, prawda?

Czy po prostu nie rozumiem związku obserwatora/obserwowalnego?

Answer 1

W RX, Observer and Observable to odrębne podmioty. Obserwator subskrybuje Obserwowalny. Obserwowalne emituje przedmioty do obserwatorów, wywołując metody obserwatorów. Jeśli chcesz wywołać metody obserwatora spoza zakresu Observable.create(), możesz użyć Tematu, który jest proxy, który działa jako obserwator i jest obserwowalny w tym samym czasie.

Można zrobić tak:

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

można znaleźć więcej informacji na temat podmiotów tutaj:

• https://github.com/Reactive-Extensions/RxJS/blob/master/doc/api/subjects/subject.md
• http://reactivex.io/documentation/subject.html

Answer 2

wierzę Observable.create() nie przyjmuje obserwatora jako parametru zwrotnego, ale emitera. Więc jeśli chcesz dodać nową wartość do Obserwowalne spróbować zamiast:

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" 

Tak Temat ułatwia, zapewniając Obserwowalne i obserwatorem tego samego obiektu, ale nie jest to dokładnie taka sama, jak Przedmiot pozwala na zasubskrybować wielu obserwatorów do tego samego obserwowalnego, gdy obserwowalne dane wysyłają tylko do ostatniego obserwowanego obserwatora, więc używajcie go świadomie. Oto JsBin, jeśli chcesz majstrować przy nim.