Как я могу подписаться и отказаться от подписки на индивидуальные запросы пожарного магазина?

Question

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

@Effect()
startStopQuery$ = this.actions$.pipe(
  ofType(
    ActionTypes.START,
    ActionTypes.STOP
  ),
  switchMap(action => {
    if (action.type === ActionTypes.STOP) {
      return of([]);
    } else {
      return this.afs.collection('collection', ref => {
        return ref.where('field', '==', 'x');
      }).stateChanges();
    }
  }),
  mergeMap(actions => actions),
  map(action => {
    return {
      type: `[Collection] ${action.type}`,
      payload: { id: action.payload.doc.id, ...action.payload.doc.data() }
    };
  })
);

Что я на самом деле хочу сделать, так это иметь несколько текущих запросов, чтобы я мог запускать и останавливать те же два действия, но там, где это зависит от полезной нагрузки действия. Когда я изменял его каждый раз, когда выполнял новый запрос, последний перестал работать. Я думаю, это потому, что оператор switchMap переключается с моего последнего наблюдаемого запроса. Это лучшее, что я придумал:

@Effect()
startStopQueryById$ = this.actions$.pipe(
  ofType(
    ActionTypes.START_BY_ID,
    ActionTypes.STOP_BY_ID
  ),
  switchMap(action => {
    if (action.type === ActionTypes.STOP_BY_ID) {
      return of([]);
    } else {
      return this.afs.collection('collection', ref => {
        return ref.where('field', '==', action.id);
      }).stateChanges();
    }
  }),
  mergeMap(actions => actions),
  map(action => {
    return {
      type: `[Collection] ${action.type}`,
      payload: { id: action.payload.doc.id, ...action.payload.doc.data() }
    };
  })
);

Как я уже сказал, я думаю, что проблема в операторе switchMap. Но это также то, от чего я зависел, чтобы заставить «стоп» работать в первую очередь. Кажется, я не могу найти другое решение, так как еще не очень хорошо разбираюсь в стиле.

Любая помощь будет принята с благодарностью!

Answer 1

Я придумал решение. Я делаю объект, который сопоставляет идентификаторы с наблюдаемыми объектами изменений в хранилище. На начальном действии я делаю слушателя и добавляю его к объекту. Я удостоверяюсь, что он автоматически отписался, отправив takeUntil с соответствующим действием остановки. Он возвращает merge всех наблюдаемых в объекте, и я делаю глупые действия, как и раньше. У меня также есть отдельный эффект, вызванный действием остановки, чтобы удалить наблюдаемое с объекта. Это выглядит так:

queriesById: {[id: string]: Observable<DocumentChangeAction<Element>[]>} = {};
@Effect()
startQuery$ = this.actions$.pipe(
  ofType(ActionTypes.START_BY_ID),
  switchMap(action => {
    this.queriesByPlay[action.pid] = this.afs.collection<Element>('requests', ref => {
      return ref.where('field', '==', action.id);
    }).stateChanges().pipe(
      takeUntil(
        this.actions$.pipe(
          ofType(ActionTypes.STOP_BY_ID),
          filter(cancelAction => action.id === cancelAction.id),
        )
      )
    );
    return merge(
      Object.values(this.queriesByPlay)
    );
  }),
  mergeMap(actions => actions),
  mergeMap(actions => actions),
  map(action => {
    return {
      type: `[Collection] ${action.type}`,
      payload: { id: action.payload.doc.id, ...action.payload.doc.data() }
    };
  })
);

Effect({dispatch: false})
stopQuery$ = this.actions$.pipe(
  ofType(ActionTypes.STOP_BY_ID),
  map(action => delete this.queriesByPlay[action.id]),
);

Кажется, что это работает и не имеет никаких проблем, кроме сложного понимания.