RXJS - uruchamiać timer tylko wtedy, gdy nie jest używany?

Question

Używam strumienia, który jest dławiony po przewinięciu okna.
Podczas dławienia (tak długo, jak przewijanie), wysyła wartości do konsoli.

Jednak, gdy strumień jest bezczynny (użytkownik jest nie przewijanie okna) - Chcę, aby zegar do kick. Jednak - jeśli użytkownik ponownie przewija - nie chcę, aby timer emitował wartości.

Obecnie robię tak:

const observable = Rx.Observable.fromEvent(window, 'scroll'); 

    const subscriber = observable 
     .throttleTime(300) 
     .map(() => 'throttle') 
     .merge(Rx.Observable.interval(1000).map(() => 'tick')) 
     .subscribe(
      (x) => { 
      console.log('Next: event!', x); 
      }, 
      (err) => { 
      console.log('Error: %s', err); 
      }, 
     () => { 
      console.log('Completed'); 
      }); 

Problemem jest to, że podczas przewijania - Widzę zarówno "throttle" I "tick" (mam tylko zobaczyć "przepustnicy")

myśleć o tym z innego POV. Praca zawsze musi zostać uruchomiona. Jeśli przewijam - ten zwężony przewijanie - powinien wywołać zadanie. Jeśli nie przewijam - timer powinien zacząć działać. (i zatrzymuje się, gdy użytkownik ponownie zacznie przewijać).

Pytanie:
Jak mogę uruchomić stoper po bezczynności nie przewijania?

PLNKR

Answer 1

zrobiłbym to tak:

const scroll$ = Rx.Observable.fromEvent(window, 'scroll') 
    .throttleTime(300 /* ms */) 
    .publish(); 

scroll$.connect(); 

const subscriber = scroll$ 
    .map(() => 'throttle') 
    .race(Rx.Observable.interval(1000).map(() => 'tick')) 
    .take(1) 
    .repeat() 
    .subscribe(
     (x) => { 
      console.log('Next: event!', x); 
     }, 
     (err) => { 
      console.log('Error: %s', err); 
     }, 
     () => { 
      console.log('Completed'); 
     }); 

używa operatora race() subskrybować jedynie obserwowalnym który emituje pierwszy który jest 1s interval lub zdarzenie przewijania. Zaraz potem chcę zacząć od nowa z innym interwałem, więc używam take(1).repeat().

Musiałem również zmienić obserwowalny scroll$ w gorący Obserwowalny, aby utrzymać throttleTime() w działaniu powtarzających się subskrypcji.

Zaktualizowana demo: https://plnkr.co/edit/sWzSm32uoOQ1hOKigo4s?p=preview

Answer 2

Można użyć debounceTime wykryć okresy bez przewijania.

const scroll = Rx.Observable.fromEvent(window, 'scroll') 
    .throttleTime(300) 
    .mapTo(false); 
const noscroll = Rx.Observable.fromEvent(window, 'scroll') 
    .startWith(0) // init with no scroll. 
    .debounceTime(300) // detect no scroll after 300 ms. 
    .mapTo(true); 
scroll.merge(noscroll) 
    .switchMap(e => e ? Rx.Observable.interval(1000).mapTo("Tick!") : Rx.Observable.of("Scroll!")) 
    // start the interval if there was no scroll. Stop the interval if there was a scroll. 
    .subscribe(updateTimer) 

Innym problemem z kodem korzysta merge że zachowa oba źródła subskrybowanych, zamiast używać switchMap (rodzeństwo mergeMap), który będzie zapisać się do wewnętrznej obserwowalnym każdym nowym wydarzeniem jest emitowany, ale również wypisać poprzednie wewnętrzne źródło, jeśli inne wydarzenie jest emitowane ze źródła.

Re: "Kolejne POV" część pytania: Możesz zastąpić Rx.Observable.interval(1000) w switchMap z zadaniem. Przewijanie spowoduje anulowanie/anulowanie subskrypcji zadania (gdy zostanie wyemitowane empty), jeśli nie będzie już przewijania, zadanie zostanie uruchomione ponownie.

Live demo