在 RXJS 中发出结果并链接新请求

Question

我正在尝试在RXJS中实现一个超级简单的缓存。我希望它的工作方式是：

1. 首先从本地存储中提取，然后将存储在那里的值发送给订阅者。
2. 调用 Web 服务以获取存储在本地存储中的内容的最新版本。
3. 将更新的值保存回本地存储。
4. 将最新值返回给订阅者。

我尝试了以下代码，但直到对服务的调用响应后，它似乎才返回。SwitchMap也没有做我想做的事。

const partnerKey = `partner-${partnerId}`;
const localPartner = JSON.parse(localStorage.getItem(partnerKey));
this.partner$ = of(localPartner)
.pipe(concatMap(_ => partnersService.getById(partnerId)
.pipe(tap(partner => {
localStorage.setItem(partnerKey, JSON.stringify(partner));
return partner;
}))));
this.partner$.subscribe(x=> console.log(x));

在上面的示例中，console.log应该调用两次。我本以为concatMap会有多个发射，但也许我理解错了。感谢您的帮助！

Answer 1

@kyler-Johnson让我走上了正确的轨道。我错误地认为，这应该是可管道的，但事实并非如此。我需要在管道外使用操作员。在这种情况下，我最终将两个数据提取拆分为它们自己的可观察量，然后将它们连接起来。

const partnerKey = `partner-${partnerId}`;
const fetchDataFromLocalStorage$ = of(JSON.parse(localStorage.getItem('partner-' + partnerId)));
const fetchDataFromServer$ = partnersService.getById(partnerId)
.pipe(tap(partner => localStorage.setItem('partner-' + partnerId, JSON.stringify(partner))));
this.partner$ = concat(fetchDataFromLocalStorage$, fetchDataFromServer$);

Answer 2

您可以使用直接发出值并使用先前输出作为输入递归调用的expand。在第一次执行后返回EMPTY以结束递归。

const { of, EMPTY } = rxjs; 
const { delay, expand } = rxjs.operators;
const fetchLocal = of('from local').pipe(delay(500))
const fetchRemote = of('from remote').pipe(delay(3000))
const source = fetchLocal.pipe(
expand((value, index) => index < 1 ? fetchRemote : EMPTY)
);
source.subscribe(console.log);

<script src="https://unpkg.com/rxjs/bundles/rxjs.umd.min.js"></script>