Makel API-Aufruf von einem Bach

Question

Hier:

import Rx from 'rxjs'; 

function fakeApi(name, delay, response) { 
    return new Rx.Observable(observer => { 
    console.log(`${name}: Request.`) 
    let running = true; 
    const id = setTimeout(() => { 
     console.log(`${name}: Response.`) 
     running = false; 
     observer.next(response); 
     observer.complete(); 
    }, delay); 
    return() => { 
     if(running) console.log(`${name}: Cancel.`) 
     clearTimeout(id); 
    } 
    }) 
} 

function apiSearch() { return fakeApi('Search', 4000, "This is a result of the search."); } 


//============================================================ 

const messages$ = new Rx.Subject(); 

const toggle$ = messages$.filter(m => m === 'toggle'); 

const searchDone$ = toggle$.flatMap(() => 
    apiSearch().takeUntil(toggle$) 
); 

searchDone$.subscribe(m => console.log('Subscriber:', m)) 

setTimeout(() => { 
    // This one starts the API call. 
    toggle$.next('toggle'); 
}, 2000) 

setTimeout(() => { 
    // This one should only cancel the API call in progress, not to start a new one. 
    toggle$.next('toggle'); 
}, 3000) 

setTimeout(() => { 
    // And this should start a new request again... 
    toggle$.next('toggle'); 
}, 9000) 

meine Absicht ist es, den API-Aufruf zu starten und stoppen, wenn es läuft nach dem gleichen toggle$ Signal ist. Problem mit dem Code ist, dass toggle$ jedes Mal einen neuen API-Aufruf startet. Ich möchte, dass der neue Anruf nicht gestartet wird, wenn bereits einer läuft, nur um den bereits laufenden zu stoppen. Irgendwie sollte ich die äußerste flatMap von toggle$ Stream "abbestellen", während apiSearch() läuft. Ich schätze, dass es notwendig ist, den Code neu zu strukturieren, um das Verhalten zu erreichen ... Was ist der Weg von RxJS?

UPDATE: Nach weiteren Untersuchungen und Anleitung Lookups, ich kam mit dieser:

const searchDone$ = toggle$.take(1).flatMap(() => 
    apiSearch().takeUntil(toggle$) 
).repeat() 

Arbeiten, wie es sollte. Fühlt sich immer noch kryptisch an. Ist das, wie Sie RxJS Jungs es lösen würden?

Answer 1

Ich denke, dass Ihre Lösung nur einmal funktioniert, seit Sie take(1) verwenden. Man könnte es wie folgt zu tun:

const searchDone$ = toggle$ 
    .let(observable => { 
     let pending; 

     return observable 
      .switchMap(() => { 
       let innerObs; 

       if (pending) { 
        innerObs = Observable.empty(); 
       } else { 
        pending = innerObs = apiSearch(); 
       } 

       return innerObs.finally(() => pending = null); 
      }); 
    }); 

I let() nur pending zu wickeln, ohne es zu Umfang in Mutter erklärt bin mit. Der Operator switchMap() kündigt für Sie automatisch ab, ohne take*() zu verwenden.

Die Ausgabe mit Ihrem Test setTimeout s wie folgt sein:

Search: Request. 
Search: Cancel. 
Search: Request. 
Search: Response. 
Subscriber: This is a result of the search.