Как сделать паузу и запуск intrerval в RxJS?

Question

[Programep]

Добрый день!
Подскажите, пожалуйста, как решить такую задачу. Также просьба посоветуйте руководства по RxJS с простыми примерами.

1. Проект разрабатываю на Angular 7.0.4 и rxjs 6.3.3
   
2. на странице есть две подписки на два interval.
   1-й: interval(10000); внутри subscribe этого interval отрабатывают http-запросы.
   2-й: interval(1000); этот интервал изменяет переменную second_to_event на -1. При выполнении условия second_to_event == 0, нужно:
   
   1. приостановить эту подписку
      
   2. выполнить http-запрос
      
   3. в случае успешного выполнения http-запрос, сбросить переменную second_to_event, запустить подписку
   
   

Через google нашел, что для этого используют switchMap: вот пример. Но в этом примере Observable создается на событие, а в моем случае, как я понимаю, нужно сделать Observable на условие.
Вот код, который у меня сейчас есть:

export class GameComponent implements OnInit, OnDestroy {

  reloadGameInfo = interval(10000);
  reloadGameInfoSubscription: Subscription;
  stepTimer = interval(1000);
  stepTimerSubscription: Subscription;
  
  ngOnInit(){
	  this.reloadGameInfoSubscription = this.reloadGameInfo.subscribe(() => {      
      this.gameService.getInfoGame(this.current_game.auth_key)
        .subscribe((response: Game) => {          
          if (this.current_game.status_id == gameStatusCode.two_player_in_game) {            
            if (!this.stepTimerSubscription) {
              this.stepTimerSubscription = this.stepTimer.subscribe(() => {                                
                if (--this.current_game.seconds_for_step == 0) {
                  this.stepTimerSubscription.unsubscribe();                  
                }
              });
            }
          }
        });
    });
  }
}

Answer 1

[Антон Швец]

reloadGameInfo = interval(10000);
stepTimer = interval(1000);
second_to_event = 100;

this.stepTimer.subscribe(() => {
  this.second_to_event = this.second_to_event >= 0 ? this.second_to_event - 1 : 0;
  if (this.second_to_event === 0) {
    // make request at second_to_event equals to 0 and set second_to_event if request success
  }
});

this.reloadGameInfo.pipe(
  filter(() => this.second_to_event > 0
)
  .subscribe(() => {
    // do 10000 periodic task
  })

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

Comments

[Programep]

Спасибо! Только сделал немного по-другому (в вашем примере reloadGameInfo будет стоят до тех пор пока this.second_to_event не будет снова больше 0, но его останавливать не нужно было, а нужно останавливать stepTimer):

reloadGameInfo = interval(10000);
stepTimer = interval(1000);
second_to_event = 100;
this.reloadGameInfoSubscription = this.reloadGameInfo.subscribe(() => {      
      this.gameService.getInfoGame(this.current_game.auth_key)
        .subscribe((response: Game) => {          
          /* update game info */
        });
    });
    this.stepTimerSubscription = this.stepTimer
      .pipe(filter(() => this.current_game.current_seconds_for_step > 0))
      .subscribe(() => {
        if (--this.current_game.current_seconds_for_step == 0) {
          this.gameService.skipStep(this.current_game.auth_key).subscribe(
            (response: any) => {
              if (response.success) {
                this.current_game.current_seconds_for_step = this.current_game.default_seconds_for_step;
              }
            }
          );
        }
      });

Помогите, пожалуйста, разобраться еще в нескольких вопросах.
1. Скажите, пожалуйста, правильно ли я понимаю что pipe с RxJS 5.5 добавили
для возможности замены такого кода
of(1,2,3).map(x => x + 1).filter(x => x > 2);
на такой:

of(1,2,3).pipe(
  map(x => x + 1),
  filter(x => x > 2)
);

? Или это сделано для другого?
2. Я просмотрел статью Укрощение Змейки с помощью реактивных потоков - увидел там пример что можно объединять подписки с помощью оператора combineLatest. Получается мне нужно убрать .subscribe() у this.stepTimer, this.reloadGameInfo и объединить их с помощью combineLatest, или как вы говорили flatMap(он же mergeMap)? Но не пойму где тогда указать то, что сейчас делается внутри subscribe.
3. Как вы думаете, какой лучше оператор использовать mergeMap/flatMap или combineLatest для объединения stepTimer и reloadGameInfo? Из документации понял что разница между mergeMap/flatMap и combineLatest, в том что при combineLatest будет содержать последние данные из stepTimer И reloadGameInfo, а при mergeMap/flatMap будут данные либо из stepTimer, либо из reloadGameInfo.

[Антон Швец]

Programep,
1. pipe для возможности импортировать операторы по отдельности как простые функции и добавлять свои операторы без манкипатчинга прототипа Observable. Раньше это был let
2. mergeMap сливает наш поток с потоком из другого обсервабла.
combineLatest делает новый поток, состоящий из массивов последних значений входящих потоков. И срабатывает каждый раз, когда приходит новое значение одного из составляющих.

[Programep]

Антон Швец, спасибо! Разобрался теперь как это работает. Подскажите, пожалуйста, а как можно объединить два interval в одну подписку, чтобы не делать unsubscribe на для каждого inverval? Пробовал через mergeMap, но получается создаются лишние interval.

[Антон Швец]

Programep, interval это порождающий оператор, а не подписка.
Вы хотите это чтоль?

import { interval, merge } from 'rxjs';
merge(
  interval(300),
  interval(1000),
).subcribe(console.log)

Или объединять подписки?

const sub = new Subscription();
const s1 = interval(300).subcribe(console.log);
const s2 = interval(1000).subcribe(console.log);
sub.add(s1).add(s2)
setTimeout(() => sub.unsubscribe(), 10000);

[Programep]

Антон Швец, не совсем понял в чем разница двух решений. Как понимаю в первом варианте будет по сути одна подписка, а во втором две объединены в одну. По сути получается один и тот же результат. Сейчас у меня два таймера: 1-й таймер обновляет информацию от сервера, а 2-й таймер обновляет счетчик обратного отсчета времени. Посмотрев пример из статью Укрощение Змейки с помощью реактивных потоков , подумал может быть имеет смысл объединить потоки в какой-то один глобальный, чтобы при необходимости можно остановить только глобальный поток, а не писать unsubscribe для каждого потока. Сейчас у меня сделано так, может быть можно улучшить?

this.reloadGameInfoSubscription = this.reloadGameInfo$
      .pipe(
        mergeMap((x) => this.gameService.getInfoGame(this.current_game.auth_key)
          .pipe(
            map((game: Game) => {
              return { tick: x, game: game };
            })
          )
        )
      ).subscribe((response: { tick: number, game: Game }) => {
        /*
        логика обработки
        */
      });
    this.stepTimerSubscription = this.stepTimer$
      .pipe(
        filter(() => this.current_game.current_seconds_for_step > 0
          && this.field.start_word.length > 0
          && this.current_game.status_id == gameStatusCode.two_player_in_game),
      ).subscribe(() => {
        if (--this.current_game.current_seconds_for_step == 0) {
          this.gameService.skipStep(this.current_game.auth_key)
            .pipe(
              mergeMap((skipStepResponse) => this.gameService.getInfoGame(this.current_game.auth_key)
                .pipe(
                  map((game: Game) => {
                    return { skipStepResponse: skipStepResponse, game: game };
                  }))
              )
            )
            .subscribe(
              (response) => {
                if (response.skipStepResponse.success) {
                  this.current_game.current_seconds_for_step = this.current_game.default_seconds_for_step;
                  this.gameService.setInfoGame(response.game);
                }
              }
            );
        }
      });

И еще один вопрос, правильно ли я понимаю что filter фактически не делает никакую реальную паузу, а только не дает выполнится коду который написан в subscribe() в примере ниже:

this.stepTimer$
      .pipe(
        filter(() => this.current_game.current_seconds_for_step > 0
          && this.field.start_word.length > 0
          && this.current_game.status_id == gameStatusCode.two_player_in_game),
      ).subscribe();

[Антон Швец]

Programep,

не совсем понял в чем разница двух решений.

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

А filter это фильтр, фильтрует поток по условию, не понял какая тут может быть пауза.

[Programep]

Антон Швец,

Создание нового потока путем слияния двух интервалов, и складирование подписок в одну для удобной отписки всех скопом.

Но при слиянии потов ведь тоже будет одна подписка, только, как я понимаю, логику придется писать в одном subscribe вместо нескольких, как это работает сейчас.

А filter это фильтр, фильтрует поток по условию, не понял какая тут может быть пауза.

Изначально я хотел сделать что-то похожее на это:

if (need_start_timer){
timer_step.start();
} else {
timer_step.pause();
}