làm thế nào có thể bắt đầu và dừng một khoảng thời gian quan sát được trong RXJS?

Question

Tôi có một bộ đếm thời gian rất đơn giản có thể quan sát được và tôi muốn bắt đầu/dừng truyền mà không ngắt kết nối thuê bao (mà phải ngồi và đợi bất kể trạng thái quan sát được). Có thể, và nếu như vậy làm thế nào?

var source = Rx.Observable 
    .interval(500) 
    .timeInterval() 
    .map(function (x) { return x.value + ':' + x.interval; }) 
    .take(10); 

    var subscription = source.subscribe(
    function (x) { 
    $("#result").append('Next: ' + x + ' '); 
    }, 
    function (err) { 
    $("#result").append('Error: ' + err); 
    }, 
    function() { 
    $("#result").append('Completed'); 
    }); 

nhận xét chung: hầu hết các ví dụ ive đều cho thấy cách xác định quan sát và người đăng ký. làm thế nào để tôi ảnh hưởng đến hành vi của các đối tượng hiện có?

Answer 1

Phụ thuộc vào nguồn tín hiệu dừng/tiếp tục là gì. Cách đơn giản nhất mà tôi có thể nghĩ đến là với pausable operator, vì tài liệu này cho biết hoạt động tốt hơn với các quan sát nóng. Vì vậy, trong đoạn mã mẫu sau, tôi đã xóa take(10) (tín hiệu tạm dừng của bạn hiện đã đi qua chủ đề pauser) và thêm share để biến thiết bị của bạn thành một thiết bị nóng.

• Về nóng và lạnh, hãy xem illustrated respective data flows.
• Mở đối tượng, bạn cũng có thể xem lại corresponding semantics

var pauser = new Rx.Subject(); 
var source = Rx.Observable 
    .interval(500) 
    .timeInterval() 
    .map(function (x) { return x.value + ':' + x.interval; }) 
    .share() 
    .pausable(pauser); 

var subscription = source.subscribe(
    function (x) { 
    $("#result").append('Next: ' + x + ' '); 
    }, 
    function (err) { 
    $("#result").append('Error: ' + err); 
    }, 
    function() { 
    $("#result").append('Completed'); 
}); 

    // To begin the flow 
pauser.onNext(true); // or source.resume(); 

// To pause the flow at any point 
pauser.onNext(false); // or source.pause(); 

Đây là một more sophisticated example đó sẽ tạm dừng nguồn của bạn mỗi 10 hạng mục:

// Helper functions 
function emits (who, who_) {return function (x) { 
who.innerHTML = [who.innerHTML, who_ + " emits " + JSON.stringify(x)].join("\n"); 
};} 

var pauser = new Rx.Subject(); 
var source = Rx.Observable 
    .interval(500) 
    .timeInterval() 
    .map(function (x) { return x.value + ':' + x.interval; }) 
    .share(); 
var pausableSource = source 
    .pausable(pauser); 

source 
    .scan(function (acc, _){return acc+1}, 0) 
    .map(function(counter){return !!(parseInt(counter/10) % 2)}) 
    .do(emits(ta_validation, 'scan')) 
    .subscribe(pauser); 

var subscription = pausableSource.subscribe(
    function (x) { 
    $("#ta_result").append('Next: ' + x + ' '); 
    }, 
    function (err) { 
    $("#ta_result").append('Error: ' + err); 
    }, 
    function() { 
    $("#ta_result").append('Completed'); 
}); 

Bạn sẽ trông như thế câu trả lời của bạn cho câu hỏi thứ hai. Kết hợp các quan sát bạn được cung cấp với các nhà khai thác RxJS có liên quan để nhận ra trường hợp sử dụng của bạn. Đây là những gì tôi đã làm ở đây.