反应式扩展中的数据服务

Question

我想要一个通用类来处理数据处理创建、更新和删除的内存缓存。底层模型继承自一个 Id 类型为字符串的接口。

interface IModel
{
  string Id { get; }
}

处理创建和更新很容易。例如，如果我想订阅流并填充字典，我知道如果模型 ID 不存在则需要创建，否则它是更新。

我的问题是：
您将如何处理删除而不引入另一个类来包装我的模型？我想保留IObservable<TModel>，而不是IObservable<Event<TModel>> 或IObservable<Pair<string, TModel>>，但我不知道怎么做。这可能吗？

interface IDataService<TModel>
{
  IObservable<TModel> DataStream { get; }
}

Answer 1

正如@Enigmativity 所建议的，您可以使用嵌套的可观察序列来解决这个问题。 IntroToRx 的 Sequences of coincidence 部分提到了这一点。

这将如何运作？

您可以将嵌套序列想象成二维数组，或者更具体地说是jagged array。 外部序列是内部序列的容器。 内部序列的到来代表了一个模型的创建。

interface IDataService<TModel>
{
    IObservable<IObservable<TModel>> DataStream { get; }
}

一旦你有了一个内部序列，它产生的所有值都是更新的（第一个除外）。 内部序列只会为单个 id 生成更新。 当内部序列完成时，代表一个删除。

此模式适用于各种用例，如上面链接的第一段所述。

作为一个大理石图，您将拥有如下所示的内容。 每行代表一个内部序列。

m1  1---2----3--|
m2     a----|
m3       x----y---z--

这将导致以下逻辑流程：

1. 创建状态为 '1' 的 m1
2. 创建状态为“a”的 m2
3. 使用值“2”更新 m1
4. 创建值为“x”的 m3
5. 删除平方米
6. 使用值“3”更新 m1
7. 使用值“y”更新 m3
8. 删除m1
9. 使用值“z”更新 m3