如何使用 RXJS 刷新陈旧但活跃的数据答案

【问题标题】：How to use RXJS to refresh stale but active data如何使用 RXJS 刷新陈旧但活跃的数据
【发布时间】：2019-03-03 22:57:28
【问题描述】：

我有一组 observables 作为数据存储，我想确保那些热的不会过时。例如，如果没有可观察的订阅者，那么我不需要更新它，但如果有订阅者，那么它不应该超过 5 分钟没有更新。

timeout 似乎是我想要的，但我不希望它出错，我只想在超时时运行代码（类似于tap）。 timeoutwith 更接近，因为我可以用 observable 替换它。但是，它似乎并不“正确”，并且在一次超时后，替换的 observable 将不会附加 timeoutWidth 运算符。

function getData(index){
    return database[index].pipe(timeoutWith(5*60*1000, database[index]))
}

我刚刚想到的一种方法是添加递归addTimeout 方法，但这感觉很hacky：

private addTimeout(obs:Observable<any>, time:number):Observable<any>{
    return obs.pipe(timeoutWith(time,this.addTimeout(obs,time)));
  }

编辑，使用超时/重试添加代码：

const staleDataTime = 5*60*1000; // 5 minutes


// bhs makes it so you can update the value manually whenever you want
let bhs = new BehaviorSubject<Summary>(initialSummary);

// This will reset bhs after the timeout
bhs.pipe(timeout(staleDataTime)).subscribe(success=>null,err=>bhs.next(null));


let obs = bhs.pipe(
   switchMap(val=>val?of(val):this.retreiveSummary(filter)),
   timeout(staleDataTime),
   retry(),
   distinctUntilChanged((a:Summary,b:Summary) => a.count==b.count && a.sum == b.sum),      
   share()
);

obs.subscribe(val=>console.log("obs val: ", val));

【问题讨论】：

标签： angular rxjs

【解决方案1】：

您甚至不需要自定义运算符，因为您可以使用 timeout() 和 retry()：

return database[index].pipe(
  timeoutWith(5 * 60 * 1000),
  retry(),
);

retry() 如果发出错误，将重新订阅其源。但是，这一切都希望database[index] 在您订阅它时始终刷新。

【讨论】：

我花了很长时间，但我最终能够让它工作。用我的代码更新了我的问题

【解决方案2】：

我意识到这对自定义运算符来说并不难

import { pipe } from 'rxjs';
import { tap } from 'rxjs/operators';


export const tapTimeout = (onTimeout:(()=>any), delay:number) => {
    let timeout = window.setTimeout(onTimeout, delay);
    return pipe(tap(x=> {
        window.clearTimeout(timeout);
        timeout = window.setTimeout(onTimeout,delay);
    }));
}

【讨论】：

【解决方案3】：

有一种 RX 操作员的方式来做到这一点

obs.pipe(
    switchMap(()=>timer(5*60*1000).pipe(
       tap(_=>..dosomething),
       repeat())
  )
)

【讨论】：