RxJS: Làm cách nào để "cập nhật thủ công" một Quan sát được?

Question

Tôi nghĩ tôi phải hiểu lầm điều gì đó cơ bản, bởi vì trong tâm trí của tôi, đây là trường hợp cơ bản nhất để có thể quan sát được, nhưng với cuộc sống của tôi, tôi không thể tìm ra cách để làm điều đó từ các tài liệu.

Về cơ bản, tôi muốn để có thể làm điều này:

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

Nhưng tôi đã không thể tìm thấy một phương pháp như push. Tôi đang sử dụng điều này cho một trình xử lý nhấp chuột và tôi biết rằng họ có Observable.fromEvent cho điều đó, nhưng tôi đang cố gắng sử dụng nó với React và tôi muốn có thể chỉ cập nhật dữ liệu trong một cuộc gọi lại, thay vì sử dụng hoàn toàn hệ thống xử lý sự kiện khác nhau. Vì vậy, về cơ bản tôi muốn điều này:

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

Gần nhất tôi đã được sử dụng observer.onNext('foo'), nhưng điều đó dường như không thực sự làm việc và đó là kêu gọi các quan sát viên, trong đó có vẻ không đúng. Người quan sát phải là thứ phản ứng với luồng dữ liệu, chứ không phải thay đổi nó, đúng không?

Tôi có không hiểu mối quan hệ người quan sát/quan sát không?

Answer 1

Trong RX, Observer và Observable là các thực thể riêng biệt. Một người quan sát đăng ký một Observable. An Observable phát ra các vật phẩm cho các quan sát viên của nó bằng cách gọi các phương pháp của người quan sát. Nếu bạn cần gọi các phương pháp quan sát bên ngoài phạm vi Observable.create(), bạn có thể sử dụng Chủ đề, là một proxy hoạt động như một người quan sát và có thể quan sát cùng một lúc.

Bạn có thể làm như thế này:

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

Bạn có thể tìm thêm thông tin về các đối tượng ở đây:

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

Answer 2

Tôi tin Observable.create() không coi người quan sát là tham số gọi lại mà là người phát. Vì vậy, nếu bạn muốn thêm một giá trị mới cho Quan sát bạn thử loại này thay vì:

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" 

Có Chủ đề làm cho nó dễ dàng hơn, cung cấp Quan sát và Observer trong cùng một đối tượng, nhưng nó không phải là giống hệt nhau, như Chủ đề cho phép bạn đăng ký nhiều người quan sát đến cùng một quan sát khi một người quan sát chỉ gửi dữ liệu đến người quan sát đã đăng ký lần cuối, vì vậy hãy sử dụng nó một cách có ý thức. Đây là JsBin nếu bạn muốn tinker với nó.