javascript - 使用时初始化缓存

Question

假设我有以下事件:

• 做点什么
• 取东西
• 取得成功

DoSomething 做一些需要缓存数据的事情。当我触发事件时，我想查看缓存并对其进行处理(如果存在)。如果没有，那么我想获取它，等待它进入缓存，然后重试。

我提出了以下解决方案，但感觉我在滥用 map 运算符。 是否有更合适的操作符来观察流并抛出错误或达到重试的效果？

const DO_SOMETHING = 'DO_SOMETHING';
const FETCH_SOMETHING = 'FETCH_SOMETHING';
const FETCH_SOMETHING_SUCCESSFUL = 'FETCH_SOMETHING_SUCCESSFUL';

const events = new Rx.Subject();
const cache = new Rx.BehaviorSubject(null);

// the magic sauce
const cacheInitializer = cache
// is there a better way than this?
.map(x => {
  if (!x) { throw 'empty'; }
  return x;
})
.retryWhen(x => {
  console.log('Cache is empty');
  events.next(FETCH_SOMETHING);
  return events.filter(x => x === FETCH_SOMETHING_SUCCESSFUL)
    .first();
});

// fake getting the data
events.filter(x => x === FETCH_SOMETHING)
.do(() => { console.log('Fetching data'); })
.delay(1000)
.subscribe(x => {
  console.log('Data fetched and put into cache');
  cache.next({ data: 1 });
  events.next(FETCH_SOMETHING_SUCCESSFUL);
});

// handle doing something
events.filter(x => x === DO_SOMETHING)
.do(() => { console.log('Check the cache'); })
.switchMapTo(cacheInitializer)
.subscribe(x => {
  	console.log('Do something', x);
});

events.next(DO_SOMETHING);

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.5.6/Rx.min.js"></script>

上下文:我正在使用 ngrx 和效果。 FETCH_SOMETHING 会触发另一个效果，FETCH_SOMETHING_SUCCESSFUL 会指示效果成功完成。在这种情况下，我的缓存是 ngrx。我不认为我会想要在 API 层中缓存，因为这仍然会导致我从缓存的响应中更新状态，而不是仅仅依赖状态中的数据。

Answer 1

你想要一个简单的shareReplay，但你需要 rxjs 5.5.0 或更高版本，因为有一个错误已被修复。

const cached$ = request$.shareReplay(1);

这将在第一次订阅时触发请求，但后续订阅将使用缓存值。

错误被传递给订阅者，然后内部主题被销毁，这样错误本身就不会被缓存。这使得 observable 可以重试。所以你可以附加任何你想要的重试逻辑(比如重试直到成功)。

最后，如果 refCount 在某个时候变为 0，缓存也会持续存在。

shareReplay 也有第二个参数，很像 ReplaySubject 构造函数，定义一个时间窗口来保留缓存。

// Faked request which sometimes errors
const request$ = Rx.Observable
  .defer(() => Rx.Observable.of(Math.random()))
  .do(() => console.log('API called'))
  .map(val => {
    if (val <= 0.3) {
      console.log('API error');
      throw val;
    } else {
      return val;
    }
  })
  .delay(250);

const cached$ = request$.shareReplay(1);

Rx.Observable.timer(0, 1000)
  .take(5)
  .switchMap(() => cached$.retry(5))
  .subscribe(console.log);

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.5.6/Rx.min.js"></script>