Как мне упорядочить действия в RxJS / redux-observable vs redux-saga? - PullRequest
Новая
       100

Как мне упорядочить действия в RxJS / redux-observable vs redux-saga?

0 голосов
/ 27 марта 2020

Я начал глубоко изучать Rx Js, одна из причин - освоить redux-observable подход к побочным эффектам, хотя я считаю, что саги более удобны и «декларативны». Я уже выучил merge/flat/concat/switchMap операторов, но это не помогло мне понять, как упорядочить вещи в rx js.

Вот пример того, что я подразумеваю под "секвенированием", например приложения Timer, где запуск может быть запланирован через некоторое время, реализованный с помощью redux-saga: 

export function* timerSaga() {
  while (true) {
    yield take('START');

    const { startDelay } = yield select(); // scheduled delay

    const [cancelled] = yield race([
      take('CANCEL_START'),
      delay(startDelay)
    ]);

    if (!cancelled) {
      yield race([
        call(function*() {
           while (true) {
             yield delay(10);
             yield put({ type: 'TICK' });
           }
        }),
        take(['STOP', 'RESET']
      ]);
    }
  }
}

. Я нахожу этот пример очень логически последовательным и понятным. Я понятия не имею, как реализовать это с помощью Redux-Observable. Пожалуйста, просто дайте мне код, который воспроизводит те же логики c, но с rxjs операторами.

Ответы [ 2 ]

0 голосов
/ 27 марта 2020

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

Генераторы удовлетворяют итератору и итерируемым протоколам, которые включают извлечение значения / события (в данном случае действия Redux) из источника и блокирование выполнения до тех пор, пока эти события не поступят.

Наблюдаемые значения pu sh, а не pull. Мы описываем и называем потоки интересующих нас событий, а затем подписываемся на них. Блокирующих вызовов нет, потому что весь наш код вызывается событиями, когда они происходят.

Этот код повторяет поведение в примере саги. 

import { interval, timer } from 'rxjs';
import { withLatestFrom, mapTo, exhaustMap, takeUntil } from 'rxjs/operators';
import { ofType } from 'redux-observable';

const myEpic = (action$, state$) => {
  // A stream of all the "cancel start" actions
  const cancelStart$ = action$.pipe(ofType('CANCEL_START'));

  // This observable will emit delayed start events that are not cancelled.
  const delayedCancellableStarts$ = action$.pipe(
    // When a start action occurs...
    ofType('START'), 

    // Grab the latest start delay value from state...
    withLatestFrom(state$, (_, { startDelay }) => startDelay),

    exhaustMap(
      // ...and emit an event after our delay, unless our cancel stream
      // emits first, then do nothing until the next start event arrives.

      // exhaustMap means we ignore all other start events while we handle
      // this one.
      (startDelay) => timer(startDelay).pipe(takeUntil(cancelStart$))
    )
  );

  // On subscribe, emit a tick action every 10ms
  const tick$ = interval(10).pipe(mapTo({ type: 'TICK' }));

  // On subscribe, emit only STOP or RESET actions
  const stopTick$ = action$.pipe(ofType('STOP', 'RESET'));

  // When a start event arrives, start ticking until we get a message to
  // stop. Ignore all start events until we stop ticking.
  return delayedCancellableStarts$.pipe(
    exhaustMap(() => tick$.pipe(takeUntil(stopTick$)))
  );
};

Важно даже если мы создаем и называем эти наблюдаемые потоки, их поведение лениво - ни один из них не «активируется» до тех пор, пока не будет подписан, и это происходит, когда вы предоставляете эту функцию epi c промежуточному программному обеспечению redux-observable.

0 голосов
/ 27 марта 2020

Я предполагаю, take() возвращает наблюдаемое, не проверял код. Вероятно, он может быть преобразован в rx fashion, как показано ниже.

Ключ здесь repeat() и takeUntil()

// outter condition for starting ticker
forkJoin(take('START'), select())
    .pipe(
        switchMap(([, startDelay]) =>
            // inner looping ticker
            timer(10).pipe(switchMap(_ => put({type: 'TICK'})), repeat(),
                takeUntil(race(
                    take('CANCEL_START'),
                    delay(startDelay)
                ))
            )
            /////////////////////
        )
    )
...