重用数据流一段时间

Question

假设有一个 API 接受查询并返回结果流，因为某些结果可能会发生变化。

type Query = { 
  species?: "dog" | "cat" | "rat", 
  name?: "string",
  status?: "lost" | "found"
}
type Result = { species: string, name: string, status: string }[]

假设有多个组件向此 API 传递查询，其中一些可能是相同的。不想向服务器发送不必要的请求并喜欢优化 - 为了做到这一点，可以封装 API 并拦截调用。

interface ServiceApi {
  request(query: Query): Observable<Result>
}

class WrappedServiceApi implements ServiceApi {
  constructor(private service: ServiceApi) { }

  request(query: Query): Observable<Result> {
    // intercepted
    return this.service.request(query);
  }
}

但是如何使用 RxJS 5 进行这种优化呢？

在 RxJS 周围做这件事可能看起来像这样：

class WrappedServiceApi implements ServiceApi {

  private activeQueries;
  constructor(private service: ServiceApi) { 
    this.activeQueries = new Map<string, Observable<Result>>();
  }

  request(query: Query): Observable<Result> {
    // it's easy to stringify query
    const hashed: string = hash(query);
    if (this.activeQueries.has(hashed)) {
      // reuse existing stream
      return this.activeQueries.get(hashed);
    } else {
      // create multicast stream that remembers last value
      const results = this.service.request(query).publishLast();
      // store stream for reuse
      this.activeQueries.set(hashed, results);
      // delete stream 5s after it closed
      results.toPromise().then(
        () => setTimeout(
          () => this.activeQueries.delete(hashed), 
          5000
        )
      );
      return results;
    }
  }
}

是否可以通过更具声明性的 rx 方式实现相同的效果？

Answer 1

我没有测试它，但我会这样做：

request(query: Query): Observable<Result> {
    return Observable.of(hash(query))
        .flatMap(hashed => {
            const activeQueries = this.activeQueries;
            if (activeQueries.has(hashed)) {
                return activeQueries.get(hashed);
            } else {
                const obs = this.service.request(query)
                    .publishReplay(1, 5000)
                    .refCount()
                    .take(1);

                activeQueries.set(hashed, obs);
                return obs;
            }
        });
}

基本上唯一的区别是我将 Observables 存储在 this.activeQueries 中，而不仅仅是它们的结果，然后我使用 .publishReplay(1, 5000) 将它们缓存 5 秒。这样，当您在 5 秒后订阅同一个 Observable 时，它​​只会重新订阅其源 Observable。