RxJs flatMapLatest/switchMap отменяет обратный вызов. Где находится onCancel()?

Ответ 1

С switchMap aka flatMapLatest вы хотите обрезать как можно быстрее выполнение текущего внутреннего потока по мере поступления новых внешних элементов. Это, безусловно, хорошее дизайнерское решение, так как в противном случае это может вызвать много путаницы и допускать некоторые призрачные действия. Если вы действительно хотите сделать что-то onCancel, вы всегда можете создать свой собственный наблюдаемый с помощью пользовательского unsubscribe обратного вызова. Но все же я бы рекомендовал не связывать unsubscribe с изменяющимся состоянием внешнего контекста. В идеале unsubscribe будет очищать только внутренние ресурсы.

Тем не менее, ваш конкретный случай можно решить без доступа к onCancel или аналогичному. Ключевое наблюдение - если я правильно понял ваш случай использования, то на search все предыдущие/ожидающие действия могут быть проигнорированы. Поэтому вместо того, чтобы беспокоиться об уменьшении счетчика, мы можем просто начать отсчет с 1.

Некоторые замечания о фрагменте:

• используется BehaviorSubject для подсчета ожидающих запросов - поскольку он готов к составлению с другими потоками;
• проверил все случаи, которые вы опубликовали в своем вопросе, и они работают;
• добавил некоторые нечеткие тесты, чтобы продемонстрировать правильность;
• Не уверен, если вы хотите разрешить nextPage, когда search все еще находится на рассмотрении, но, похоже, это просто вопрос использования concatMapTo vs merge;
• используются только стандартные операторы Rx.

PLNKR

console.clear();

const searchSub = new Rx.Subject(); // trigger search 
const nextPageSub = new Rx.Subject(); // triger nextPage
const pendingSub = new Rx.BehaviorSubject(); // counts number of pending requests

const randDurationFactory = min => max => () => Math.random() * (max - min) + min;
const randDuration = randDurationFactory(250)(750);
const addToPending = n => () => pendingSub.next(pendingSub.value + n);
const inc = addToPending(1);
const dec = addToPending(-1);

const fakeSearch = (x) => Rx.Observable.of(x)
  .do(() => console.log(`SEARCH-START: ${x}`))
  .flatMap(() => 
    Rx.Observable.timer(randDuration())
    .do(() => console.log(`SEARCH-SUCCESS: ${x}`)))

const fakeNextPage = (x) => Rx.Observable.of(x)
  .do(() => console.log(`NEXT-PAGE-START: ${x}`))
  .flatMap(() =>
    Rx.Observable.timer(randDuration())
    .do(() => console.log(`NEXT-PAGE-SUCCESS: ${x}`)))

// subscribes
searchSub
  .do(() => console.warn('NEW_SEARCH'))
  .do(() => pendingSub.next(1)) // new search -- ingore current state
  .switchMap(
    (x) => fakeSearch(x)
    .do(dec) // search ended
    .concatMapTo(nextPageSub // if you wanted to block nextPage when search still pending
      // .merge(nextPageSub // if you wanted to allow nextPage when search still pending
      .do(inc) // nexpage started
      .flatMap(fakeNextPage) // optionally switchMap
      .do(dec) // nextpage ended
    )
  ).subscribe();

pendingSub
  .filter(x => x !== undefined) // behavior-value initially not defined
  .subscribe(n => console.log('PENDING-REQUESTS', n))

// TEST
const test = () => {
    searchSub.next('s1');
    nextPageSub.next('p1');
    nextPageSub.next('p2');

    setTimeout(() => searchSub.next('s2'), 200)
  }
// test();

// FUZZY-TEST
const COUNTER_MAX = 50;
const randInterval = randDurationFactory(10)(350);
let counter = 0;
const fuzzyTest = () => {
  if (counter % 10 === 0) {
    searchSub.next('s' + counter++)
  }
  nextPageSub.next('p' + counter++);
  if (counter < COUNTER_MAX) setTimeout(fuzzyTest, randInterval());
}

fuzzyTest()

<script src="https://npmcdn.com/[email protected]/bundles/Rx.umd.js"></script>

Ответ 2

Один из способов: используйте оператор finally (rxjs4 docs, Источник rxjs5). Наконец, триггеры всякий раз, когда наблюдаемые отменяются или завершаются по любой причине.

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

var pageStream = Rx.createObservableFunction(_self, 'nextPage')
        .startWith(1)
        .concatMap(function(pageNumber) {
            ++pendingRequests;
            // assumes getPage returns an Observable and not a Promise
            return MyHTTPService.getPage(pageNumber)
               .finally(function () { --pendingRequests; })
        });

Ответ 3

Я написал решение для вашей проблемы с нуля.
Конечно, это может быть написано более функционально, но все равно работает.

Это решение основано на reqStack, которое содержит все запросы (сохраняющие порядок вызовов), где запрос является объектом с свойствами id, done и type.

Когда запрос выполняется, вызывается метод requestEnd. Есть два условия, и хотя бы одного из них достаточно, чтобы скрыть загрузчик.

• Когда последний запрос в стеке был запросом search, мы можем скрыть загрузчик.

• В противном случае все остальные запросы должны быть выполнены.

  function getInstance() {
   return {
      loaderVisible: false,
      reqStack: [],
  
      requestStart: function (req){
          console.log('%s%s req start', req.type, req.id)
          if(_.filter(this.reqStack, r => r.done == false).length > 0 && !this.loaderVisible){
              this.loaderVisible = true
              console.log('loader visible')
          }
      },
  
      requestEnd: function (req, body, delay){
          console.log('%s%s req end (took %sms), body: %s', req.type, req.id, delay, body)
          if(req === this.reqStack[this.reqStack.length-1] && req.type == 'search'){
              this.hideLoader(req)
              return true
          } else if(_.filter(this.reqStack, r => r.done == true).length == this.reqStack.length && this.loaderVisible){
              this.hideLoader(req)
              return true
          } 
          return false
      },
  
      hideLoader: function(req){
          this.loaderVisible = false
          console.log('loader hidden (after %s%s request)', req.type, req.id)
      },
  
      getPage: function (req, delay) {
          this.requestStart(req)
          return Rx.Observable
                  .fromPromise(Promise.resolve("<body>" + Math.random() + "</body>"))
                  .delay(delay)
      },
  
      search: function (id, delay){
          var req = {id: id, done: false, type: 'search'}
          this.reqStack.push(req)
          return this.getPage(req, delay).map(body => {  
                      _.find(this.reqStack, r => r.id == id && r.type == 'search').done = true
                      return this.requestEnd(req, body, delay)
                  })
      },
  
      nextPage: function (id, delay){
          var req = {id: id, done: false, type: 'nextPage'}
          this.reqStack.push(req)
          return this.getPage(req, delay).map(body => {  
                      _.find(this.reqStack, r => r.id == id && r.type == 'nextPage').done = true
                      return this.requestEnd(req, body, delay)
                  })
      },
  }
  }
  

Модульные тесты в Moca:

describe('animation loader test:', function() {

    var sut

    beforeEach(function() {
        sut = getInstance()
    })

    it('search', function (done) {
        sut.search('1', 10).subscribe(expectDidHideLoader)
        testDone(done)
    })

    it('search, nextPage', function (done) {
        sut.search('1', 50).subscribe(expectDidHideLoader)
        sut.nextPage('1', 20).subscribe(expectDidNOTHideLoader)
        testDone(done)
    })

    it('search, nextPage, nextPage', function(done) {
        sut.search('1', 50).subscribe(expectDidHideLoader)
        sut.nextPage('1', 40).subscribe(expectDidNOTHideLoader)
        sut.nextPage('2', 30).subscribe(expectDidNOTHideLoader)
        testDone(done)
    })

    it('search, nextPage, nextPage - reverse', function(done) {
        sut.search('1', 30).subscribe(expectDidNOTHideLoader)
        sut.nextPage('1', 40).subscribe(expectDidNOTHideLoader)
        sut.nextPage('2', 50).subscribe(expectDidHideLoader)
        testDone(done)
    })

    it('search, search', function (done) {
        sut.search('1', 60).subscribe(expectDidNOTHideLoader) //even if it takes more time than search2
        sut.search('2', 50).subscribe(expectDidHideLoader)
        testDone(done)
    })

    it('search, search - reverse', function (done) {
        sut.search('1', 40).subscribe(expectDidNOTHideLoader) 
        sut.search('2', 50).subscribe(expectDidHideLoader)
        testDone(done)
    })

    it('search, nextPage, search', function (done) {
        sut.search('1', 40).subscribe(expectDidNOTHideLoader) //even if it takes more time than search2
        sut.nextPage('1', 30).subscribe(expectDidNOTHideLoader) //even if it takes more time than search2
        sut.search('2', 10).subscribe(expectDidHideLoader)
        testDone(done)
    })

    it('search, nextPage (call after response from search)', function (done) {
        sut.search('1', 10).subscribe(result => {
            expectDidHideLoader(result)
            sut.nextPage('1', 10).subscribe(expectDidHideLoader)
        })
        testDone(done)   
    })

    function expectDidNOTHideLoader(result){
        expect(result).to.be.false
    }

    function expectDidHideLoader(result){
        expect(result).to.be.true
    }

    function testDone(done){
        setTimeout(function(){
            done()
        }, 200)
    }

})

Часть выхода:

JSFiddle - здесь.

Ответ 4

Я думаю, там гораздо более простое решение, чтобы объяснить это. Я хотел бы "перефразировать" примеры, которые вы дали в своем редактировании:

• Статус "ожидает" до тех пор, пока будут незаблокированные запросы.
• Ответ закрывает все предыдущие запросы.

Или, в стиле stream/marbles

(O = запрос [открытый], C = ответ [закрыть], p = ожидающий, x = не ожидающий ответа)

http stream: ------ O --- O --- O --- C --- O --- C --- O --- O --- C --- O- -

------ Статус: x ---- P -------------- x --- P ---- x ---- P ---- ----- х --- --- р

Вы можете видеть, что счетчик не имеет значения, у нас есть флаг, который на самом деле включен (ожидает) или выключен (ответ был возвращен). Это истинно из-за вас switchMap/flatMap, или, как вы сказали в конце вашего редактирования, каждый раз есть только один активный запрос.

Флаг на самом деле является булевым наблюдаемым/обервером или просто субъектом.

Итак, вам нужно сначала определить:

var hasPending: Subject<boolean> = BehaviorSubject(false);

The BehaviorSubject относится к двум причинам:

• Вы можете установить начальное значение (false = ничего не ожидается).
• Новые подписчики получают последнее значение, поэтому даже компоненты, созданные позже, будут знать, есть ли ожидающий запрос.

Чем остальное станет простым, всякий раз, когда вы отправляете запрос, установите ожидающий "true", когда запрос будет выполнен, установите флаг ожидания на "false".

var pageStream = Rx.createObservableFunction(_self, 'nextPage')
    .startWith(1)
    .do(function(pageNumber) {
        hasPending.next(true);
    })
    .concatMap(function(pageNumber) {
        return MyHTTPService.getPage(pageNumber);
    })
    .do(function(response) {
        hasPending.next(false);
    });

Rx.createObservableFunction(_self, 'search')       .flatMapLatest(функция (e) {           return pageStream;       })       .subscribe();

Это синтаксис rxjs 5, для rxjs 4 use onNext (...)

Если вам не нужна ваша квартира как наблюдаемая, просто значение, просто объявите:

var hasPending: booolean = false;

Затем в .do перед вызовом http do

hasPending = true;

и в .do после http-вызова do

hasPending = false;

И что он: -)

Btw, после повторного чтения всего, вы можете проверить это еще более простым (хотя и довольно быстро и грязным) решением: Измените сообщение http 'do' на:

.do(function(response) {
        pendingRequests = 0;
    });