Как использовать обновления реального времени Firestore (onSnapshot) с redux-observable / rxjs?

Question

Я могу настроить наблюдаемый при редуксе с обычными запросами Firestore

export const getStatementsEpic = (action$, store) => {
  return action$.ofType(GET_STATEMENTS)
    .filter(() => {
      const state = store.getState()
      return state.auth.user
    })
    .mergeMap(() => {
      console.log('action', store.getState().auth.user.uid)
      const db = firebase.firestore()
      db.settings({ timestampsInSnapshots: true })
      const query = db.collection('users')
        .doc(store.getState().auth.user.uid)
        .collection('statements')
        .orderBy('uploadedOn', 'desc')
        .limit(50)
      return query.get().then(snapshot => {
        console.log('Should have gotten snapshot')
        return getStatementsSnapshot(snapshot)
      })
    })
}

Но я хочу преобразовать это в реальное время, я попытался изменить

return query.get().then(snapshot => {

на

return query.onSnapshot(snapshot => {

Но это не работает ... Я думаю, это не обещание?Как мне решить это?

Answer 1

Вы правы, метод onSnapshot не возвращает обещание.Вместо этого он возвращает функцию, которую можно использовать для отмены подписки на уведомления об изменениях.До тех пор, пока эта функция отмены не будет вызвана, обратный вызов, передаваемый методу onSnapshot, будет вызываться при каждом изменении документа.(Документация указывает, что обратный вызов также будет вызван немедленно с текущим содержимым документа.)

Такие функции, как onSnapshot, которые используют функцию обратного вызова несколько раз, могут быть "преобразованы" в наблюдаемуюиспользуя функцию fromEventPattern.fromEventPattern принимает две функции в качестве параметров.Первая передаваемая функция должна вызвать onSnapshot, передав ей определенный RxJS-обработчик в качестве обратного вызова.Вторая передаваемая вами функция должна вызывать функцию отмены подписки, возвращаемую onSnapshot.RxJS будет вызывать первую функцию, когда вы подписываетесь на наблюдаемое (т.е. используете его в своем эпосе).RxJS вызовет вторую функцию, когда вы откажетесь от наблюдаемой.

Вот пример вашего кода, обновленного для использования fromEventPattern и новых каналов RxJS:

export const getStatementsEpic = (action$, state$) => action$.pipe(
  ofType(GET_STATEMENTS),
  withLatestFrom(state$),
  filter(([action, state]) => state.auth.user),
  mergeMap(([action, state]) => {
    const db = firebase.firestore()
    db.settings({ timestampsInSnapshots: true })
    return fromEventPattern(
      handler => db.collection('users')
        .doc(state.auth.user.uid)
        .collection('statements')
        .orderBy('uploadedOn', 'desc')
        .limit(50)
        .onSnapshot(handler),
      (handler, unsubscribe) => unsubscribe(),
    ).pipe(
      map(getStatementsSnapshot),
      takeUntil(action$.pipe(
        ofType(STOP_GET_STATEMENTS),
      )),
    )
  }),
)

Обратите внимание, что яВведен takeUntil в поток снимков.Без этого (или чего-то подобного) поток снимков никогда не закончится.Другое возможное изменение - использование switchMap вместо mergeMap.Отказ от подписки зависит только от вашего варианта использования.