RxJ испускают после задержки или следующего события в потоке

Question

У меня есть пары событий: add1 / add2 / etc и remove1 / remove2 / etc. Я хотел бы следующее:

• когда add1 испускается в потоке
  • , если DELAY происходит без новых add* выбросов
    • испускать remove1
  • если испускается add*
    • испускать remove1 для add1 немедленно
    • испускать remove* для add* после DELAY

Это должно продолжаться для всех выбросов add* в потоке.

Вот тест, который я написал с использованием теста мрамора RxJS для этого случая:

import test from 'tape'
import { set, lensPath } from 'ramda'
import { TestScheduler } from 'rxjs/testing'
import hideAfterDelay from '../another/file'
import { actionCreators } from '../another/dir'

const prefix = 'epics -> notifications'

test(`${prefix} -> hideAfterDelay`, t => {
  t.plan(1)

  const scheduler = new TestScheduler(t.deepEqual)

  const actionMap = {
    a: createAddAction('hello!'),
    b: createAddAction('goodbye!'),
    x: actionCreators.notifications.remove('hello!'),
    y: actionCreators.notifications.remove('goodbye!')
  }

  scheduler.run(({ cold, expectObservable }) => {
    const actionStream = cold('a-------a-b-a------', actionMap)
    const expected =          '-----x-----x-y----x'
    const actual = hideAfterDelay(5)(actionStream)

    expectObservable(actual).toBe(expected, actionMap)
  })
})

function createAddAction (name) {
  const action = actionCreators.notifications.add(name)
  const lens = lensPath(['payload', 'id'])

  return set(lens, name, action)
}

Я думаю тест отражает поведение, которое я описал выше и которое я хочу.

Как я могу написать это наблюдаемое? Я пытался использовать timer и race, но я не смог заставить это работать ...

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

Использование RxJS v6

Answer 1

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

Во-первых, ожидаемая мраморная диаграмма должна выглядеть следующим образом

//   input:    a-------a-b-a------
// - expected: -----x-----x-y----x
// + expected: -----x----x-y----x
//
// Note above that the middle x and y emit at the same time as new
// `add*` actions on the source stream instead of one frame later

С этим небольшим изменением, которое по-прежнему согласуется с моим описанием в вопросе, я смог пройти тест со следующим:

import { of, timer, empty } from 'rxjs'
import { switchMap, mapTo, tap, merge } from 'rxjs/operators'
import { ofType } from '../operators'
import actionTypes from '../../actionTypes/notifications'
import { actionCreators } from '../..'

export default (delay = 3000) => actionStream => {
  let immediateRemove

  return actionStream.pipe(
    ofType(actionTypes.ADD),
    switchMap(action => {
      let obs = empty()

      if (immediateRemove) {
        obs = of(immediateRemove)
      }

      const remove = actionCreators.notifications.remove(action.payload.id)

      immediateRemove = remove

      return obs.pipe(
        merge(
          timer(delay).pipe(
            tap(() => {
              immediateRemove = null
            }),
            mapTo(remove)
          )
        )
      )
    })
  )
}

Я не знаю, является ли это лучшим или правильным способом решения, но я вполне уверен, что это способ.