Jak uruchomić i zatrzymać interwał obserwowalny w RXJS?

Question

Mam bardzo prosty timeInterval obserwowalny i chcę rozpocząć/zakończyć transmisję bez odłączania abonentów (który powinien siedzieć i czekać bez względu na obserwowalny status). Jest to możliwe, a jeśli tak, to w jaki sposób?

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

Komentarz ogólny: większość przykładów, które widziałem, pokazuje, jak definiować obserwowalne i subskrybentów. jak mam wpływ na zachowanie istniejących obiektów?

Answer 1

Zależy od źródła sygnału zatrzymania/wznowienia. Najprostszy sposób, jaki mogę wymyślić, to pausable operator, który, jak mówi dokumentacja, działa lepiej z gorącymi obserwatorami. Tak więc w poniższym przykładowym kodzie usunąłem take(10) (Twój nieprzemyślany sygnał teraz przechodzi przez temat pauser) i dodano share, aby uczynić obserwowalny na gorącym.

• Informacje o gorącym kontra zimnym, spójrz na numer illustrated respective data flows.
• na tematy, można również sprawdzić 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(); 

Oto more sophisticated example który będzie wstrzymać źródła co 10 przedmiotów

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

Trzeba teraz twoja odpowiedź na drugie pytanie. Połącz obserwowalne, które otrzymałeś, z odpowiednimi operatorami RxJS, aby zrealizować swój przypadek użycia. Oto co tutaj zrobiłem.