Angular2: switchMap не отменяет предыдущие http-звонки

Я пытаюсь использовать switchMap для отмены любых предыдущих HTTP-вызовов в Angular2. Код в основном

var run = ():Observable<any> => {
        var url = 'http://...'
        return this._http.get(url)
            .map(result => {
                return var xmlData:string = result.text()

            });
    }


    function pollTasks() {
        return Observable.of(1)
            .switchMap(() => run())
            .map(res => res)
    }

    // caller can do subscription and store it as a handle:
    let tasksSubscription =
        pollTasks()
            .subscribe(data => {
                console.log('aa'+data)
            });

и поэтому я вызываю весь источник несколько раз подряд и получаю несколько ответов (например: aa + data)

Я находился под впечатлением switchMap, чтобы отменить предыдущие вызовы.

Ответ 1

Запросы должны исходить из одного и того же основного потока. Здесь функция factory, которая создаст экземпляр службы http, который должен делать то, что вы хотите:

function httpService(url) {
   // httpRequest$ stream that allows us to push new requests
   const httpRequest$ = new Rx.Subject();

   // httpResponse$ stream that allows clients of the httpService
   // to handle our responses (fetch here can be replaced with whatever
   // http library you're using).
   const httpResponse$ = httpRequest$
       .switchMap(() => fetch(url));


   // Expose a single method get() that pushes a new
   // request onto the httpRequest stream. Expose the
   // httpResponse$ stream to handle responses.
   return {
       get: () => httpRequest$.next(),
       httpResponse$
   };
}

И теперь код клиента может использовать эту услугу следующим образом:

const http = httpService('http://my.api.com/resource');

// Subscribe to any responses
http.httpResponse$.subscribe(resp => console.log(resp));

// Call http.get() a bunch of times: should only get one log!!
http.get();
http.get();
http.get();

Ответ 2

constructor(private _http:Http, private appStore:AppStore) {

        this.httpRequest$ = new Subject();


        this.httpRequest$
            .map(v=> {
                return v;
        })
        .switchMap((v:any):any => {
            console.log(v);
            if (v.id==-1||v.id=='-1')
                return 'bye, cancel all pending network calls';                
            return this._http.get('example.com)
                .map(result => {
                    var xmlData:string = result.text()
                });
        }).share()
        .subscribe(e => {                
        })
    ...

и нажать данные в:

this.httpRequest$.next({id: busId});

это отлично работает, и теперь у меня есть одна служба, через которую я могу передавать все сетевые вызовы, а также отменять предыдущие вызовы...

см. изображение ниже, по мере поступления новых вызовов, предыдущие отменены. обратите внимание, как я устанавливаю медленную сеть с задержкой в 4 сек для обеспечения того, что все работает как ожидалось...

Ответ 3

Я думаю, что когда вы используете оператор switchMap, вы можете только отменить запросы в текущем потоке данных. Я имею в виду события, которые происходят в одной и той же наблюдаемой цепочке...

Если вы несколько раз вызовете свой метод pollTasks, вы не сможете отменить предыдущие запросы, потому что они не будут в одном потоке данных... Вы создаете наблюдаемую цепочку каждый раз, когда вы вызываете метод.

Я не знаю, как вы запускаете выполнение своих запросов.

Если вы хотите выполнить свой запрос каждые 500 мс, вы можете попробовать следующее:

pollTasks() {
  return Observable.interval(500)
                .switchMap(() => run())
                .map(res => res.json());
}

В этом случае, если есть запрос на выполнение после 500 мс, они будут отменены, чтобы выполнить новый

При подходе вам просто нужно вызвать метод pollTasks.

Вы также можете инициировать цепочку асинхронной обработки на основе пользовательских событий. Например, когда символы заполняются входами:

var control = new Control();
// The control is attached to an input using the ngFormControl directive
control.valueChanges.switchMap(() => run())
                .map(res => res.json())
                .subscribe(...);

Существует предложение связать/инициировать более легкую обработку цепочки на событиях DOM (fromEvent)

Смотрите эту ссылку:

https://github.com/angular/angular/issues/4062

Ответ 4

Вы можете использовать тему, но если вы это сделаете, вам нужно управлять подпиской и публиковать. Если вы просто хотите, чтобы метод возвращал Observable, который отменял запросы в течение интервала, вот как бы я это сделал:

observer: Observer<CustomObject>;
debug = 0;

myFunction(someValue) {
    this.observable = new Observable<CustomObject>(observer => {
        if (!this.observer) {
            this.observer = observer;
        }

       // we simulate http response time
        setTimeout(() => {
            this.observer.next(someValue);
            this.debug++;
        }, 1000);
    })
    .debounceTime(3000) // We cancel request withing 3s interval and take only the last one
    .switchMap(fn)
    .do(() => {
        console.log("debug", this.debug);
    });
}

Используя одного и того же наблюдателя, давайте отменим запросы по всем желаемым (debounceTime, distinctUntilChanged,...).