将多个可观察对象合并到单个字典中

Question

我想将多个可观察对象（每个对象返回一个更新对象）组合成一个字典对象。

这是我想要达到的目标的示例：

private IObservable<IDictionary<string, IUpdate>> CreateUpdateStreams(Product product)
{
  var codeObservables = product.Codes.Select(code => CreateUpdateStream(code)).ToList();

  //??? 
  return pointObs.Merge().Select(update => ...);
}


private IObservable<IUpdate> CreateUpdateStream(string code)
{
  ...
  //return an observable of IUpdate
}

• 我想将所有 IUpdate 合并到一个单独的更新字典中，其中键 = 代码和值 = IUpdate
• CreateUpdateStreams 的调用者将知道产品，并希望根据更新对每个 Code 对象的某些属性进行更改。例如

产品 = Foo

Product.Codes = {Code1, Code2, Code3}

IDictionary = {Code1, "a"}, {Code2, "b"}, {Code3, "c"}

根据更新的值（在本例中为 a/b/c），将对相应的代码进行不同的更改，例如设置一个属性，如 Code.State = "a" 等。

由于每个 codeObservable 都会以不同的速率更新，Merge 似乎是一个明智的起点。我不确定如何让各个可观察对象的更新更新一个字典对象，该对象保留过去的值。

Answer 1

这是您的问题的一个镜头，它利用了匿名类型。它依赖于字典的副作用。请注意，由于 Rx 保证顺序行为，因此不需要在字典上进行同步。

private IObservable<IReadOnlyDictionary<string, IUpdate>> CreateUpdateStreams(Product product)
    {
        var dictionary = new Dictionary<string, IUpdate>();
         return
          product.Codes.Select(
              code => CreateUpdateStream(code).Select(update => new {Update = update, Code = code}))
              .Merge()
              .Do(element => dictionary.Add(element.Code, element.Update))
              .Select(_ => dictionary);
    }

请注意，我已将方法签名更改为返回IObservable<IReadOnlyDictionary<,>>，以防止客户端代码篡改字典。另一种选择是每次都返回字典的新副本。这确保了不可变的行为（但可能会对性能产生影响，具体取决于字典的大小），如下所示：

private IObservable<IDictionary<string, IUpdate>> CreateUpdateStreams(Product product)
    {
        var dictionary = new Dictionary<string, IUpdate>();
        return
            product.Codes.Select(
                code => CreateUpdateStream(code).Select(update => new {Update = update, Code = code}))
                .Merge()
                .Select(element =>
                {
                    dictionary.Add(element.Code, element.Update);
                    return new Dictionary<string, IUpdate>(dictionary);
                });
    }