shareReplay 直到特定条件 - 重新获取数据 - 再次使用 shareReplay，可能吗？

Question

我有一个案例，我想缓存一个 http 响应，其中包含整个应用程序所需的大量数据并且不会改变（直到现在......因此问题......），因此我的直接想法是使用shareReplay，它的工作原理是在该会话的应用程序生命周期内无需多次获取同一组数据。很简单，声明一个变量，例如 allData$ 并在所需的组件中订阅它，例如服务：

allData$ = this.getAllData().pipe(shareReplay())

但是在一个组件中...我现在可以实际修改此端点返回的数据，例如添加一个新项目，因此我需要在这些情况下重新获取数据并再次缓存该数据新的回应。有没有很好的 rxjs 方法来做到这一点？我想做这样的事情，就像在“一个班轮”中只使用上面的单个 allData$ 变量。我现在所采用的方法对我来说似乎有点难看，添加一个新的 BehaviorSubject 和一个组件可以在需要更新此数据时调用的函数，即服务代码：

allData = new BehaviorSubject({} as MyModel);
allData$ = this.allData.asObservable();

constructor(private http: HttpClient) {
  this.getAllData().subscribe();
}

// this is called by component, when the db has been modified and need to update allData
refetchData() {
  this.getAllData().subscribe();
}

getAllData() {
  return this.http.get<MyModel>('url').pipe(
    // updating the stream...
    tap(value => this.allData.next(value))
  )
}

我确定有一个不错的 rxjs 运算符可以完成所有这些操作？

Answer 1

对于此类情况，一个超级有用的运算符是扫描。它允许您从更改流中修改初始状态。流可以是任何你喜欢的，但通常我只会流函数，因为这是最灵活的。

dataChanges = new Subject<DataItem>();

allData$ = this.allData.pipe(
  switchMap(initialState => dataChanges.pipe(
    scan((state, newItem) => [...state, newItem] /* modify how you like */, initialState),
    startWith(initialState)
  )),
  shareReplay(1)
);

上面的代码获取初始数据，然后订阅流变化的可观察对象。在我的示例中，我只是添加到数组并返回它，但您基本上可以做任何您想做的事情。

正如我上面所说，您还可以传递返回更新状态的函数。这最初可能有点难以掌握，但更简洁、更灵活。

dataReducers = new Subject<(s: AllData) => AllData>();

allData$ = concat(
  this.allData.pipe(map(take(1)) // get initial data.
  this.dataReducers
).pipe(
  scan((state: AllData, reducer: (x: AllData) => AllData) => reducer(state)),
  shareReplay(1)
);

scan 的一个问题是设置初始值。在第一个示例中，我必须创建该内部流。在第二个示例中，我利用了一个事实，即如果您不提供种子，scan 将返回通过管道传递给它的第一个值。

scan 的 参数的输入如果您以第二种方式进行，则会出现问题。通过显式声明类型，我可以避免这种不稳定，但如果你不小心，这可能会导致错误。例如，如果我使用 merge 而不是 concat 并且没有使用 take(1) 运算符，那么非函数可能会作为 reducer 传递。

你可以添加一个检查来查看 reducer 是否真的是一个函数，或者你可以传递一个空对象作为种子参数，并将初始 observable 映射到一个不带参数的 reducer 函数。然而，一旦你做了这些事情，你就失去了清洁优势。出于这个原因，我更喜欢第一个示例中的表单以及第二个示例中的减速器。