RxJS: Come posso aggiornare "manualmente" un Osservabile?

Question

Penso che dovrei fraintendere qualcosa di fondamentale, perché nella mia mente questo dovrebbe essere il caso più fondamentale per un osservabile, ma per la vita del mio non riesco a capire come farlo dai documenti.

Fondamentalmente, voglio essere in grado di fare questo:

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

Ma io non sono stati in grado di trovare un metodo come push. Sto usando questo per un gestore di clic, e so che hanno Observable.fromEvent per quello, ma sto cercando di usarlo con React e preferirei essere in grado di aggiornare semplicemente il flusso di dati in un callback, invece di usare completamente diverso sistema di gestione degli eventi. Quindi, in pratica Voglio che questo:

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

Il più vicino ho ottenuto è stato utilizzando observer.onNext('foo'), ma che non sembrano funzionare in realtà e che è invitato l'osservatore, che non mi sembra giusto. L'osservatore dovrebbe essere la cosa che reagisce al flusso di dati, non cambiandolo, giusto?

Non capisco la relazione osservatore/osservabile?

Answer 1

In RX, Observer and Observable sono entità distinte. Un osservatore si iscrive a un osservabile. Un osservabile emette oggetti ai suoi osservatori chiamando i metodi degli osservatori. Se è necessario chiamare i metodi dell'osservatore al di fuori dell'ambito di Observable.create(), è possibile utilizzare un Oggetto, che è un proxy che funge da osservatore e Osservabile allo stesso tempo.

Si può fare in questo modo:

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

Potete trovare ulteriori informazioni su argomenti qui:

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

Answer 2

credo Observable.create() non accetta un osservatore come parametro di callback ma un emettitore. Quindi, se si desidera aggiungere un nuovo valore alla vostra Osservabile provare questo invece:

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" 

Sì soggetto facilita, fornendo osservabile e osservatore nello stesso oggetto, ma non è esattamente lo stesso, come soggetto che si permette di iscrivere più osservatori allo stesso osservabile quando un osservabile invia solo dati all'ultimo osservatore iscritto, quindi usarlo consapevolmente. Ecco un JsBin se vuoi armeggiare con esso.