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

Question

Представьте себе поток сообщений, каждое из которых связано с идентификатором пользователя.Для каждого входящего сообщения извлекайте связанную с ним информацию о пользователе (можно наблюдать 'user-fetch').Эти наблюдаемые пользовательской выборки останутся в живых и будут отслеживать любые будущие изменения для целевого пользователя.

Вопросы:

1. Как предотвратить создание дублирующихся наблюдаемых пользовательской выборки длязаданный идентификатор пользователя (и повторно использовать, возможно, уже созданную наблюдаемую информацию)?
2. Как правильно очистить все наблюдаемые пользователем выборки для отмены подписки и / или завершения?

Где я нахожусьat:

1. Мне не удалось определить существующий оператор или методологию предотвращения дублирования наблюдаемых, поэтому я написал оператор, аналогичный switchMap.Я не люблю этоКак это делается на практике?

2. Если я могу решить 1, я считаю, что решение для правильной очистки и повторного использования - refCount().

Answer 1

если я правильно понял проблему, у вас есть один поток, который испускает id -s и на основе событий этого потока, другой поток, который получает некоторые данные, связанные с идентификатором, из удаленного места (сервера).

Решение, которое я предлагаю, заключается в создании какого-то store для хранения кэшированных данных и после получения сообщения из потока id для его проверки и возврата либо ответа от нового запроса, либо кэшированных данных.

/** 
 * callBack end mocks an http request
 */
let callBackEnd$ = id => {
  customLog("___________________");
  customLog("Calling the server for " + id);
  customLog("___________________");

  return of({ id: id, data: `Some data about ${id}` });
};

/**
 * idStream$ mock the stream of id-s to be called trough http request
 */
let idStream$ = from([1, 2, 2, 3, 1, 5, 3, 4, 5]);

/**
 * We use reqStore$ to cache the already retrieved data
 */
let reqStore$ = new BehaviorSubject([]);

/**
 *  1. We subscribe to the message stream ( the stream that will tell us what to load )
 *
 *  2. With `latestFrom` we take the current store and check for any cached data, and return
 *  the cached data or the response of the new request
 *
 *  3. If the response of the `switchMap` doesn't exist in our store we add it.
 */
idStream$
  .pipe(
    tap(message => customLog(`Receiving command to retrieve : ${message}`)),
    withLatestFrom(reqStore$),
    switchMap(([e, store]) => {
      let elementSaved = store.find(x => x.id === e);
      return elementSaved ? of(elementSaved) : callBackEnd$(e);
    }),
    withLatestFrom(reqStore$),
    tap(([response, store]) => {
      if (!store.find(x => x.id === response.id)) {
        reqStore$.next([...store, response]);
      }
    })
  )
  .subscribe(([currentResponse, currentStore]) => {
    customLog("Receiving response for " + currentResponse.data);
  });

Вот живая демонстрация на Codesandbox Я надеюсь, что это поможет вам:)