RX Extension ToObservable() 在查询嵌套的 IColletion 时产生异常答案

【问题标题】：RX Extension ToObservable() yields an exception when querying a nested IColletionRX Extension ToObservable() 在查询嵌套的 IColletion 时产生异常
【发布时间】：2016-12-06 13:14:50
【问题描述】：

我只是想适应使用 ef 和 rx。不幸的是，我遇到了一个我无法解决的问题。我有一个名为 world.sql 的 MySQL 测试数据库。 EF产生了以下

public worldEntities()
        : base("name=worldEntities")
    {
    }

    protected override void OnModelCreating(DbModelBuilder modelBuilder)
    {
        throw new UnintentionalCodeFirstException();
    }

    public virtual DbSet<city> city { get; set; }
    public virtual DbSet<country> country { get; set; }
    public virtual DbSet<countrylanguage> countrylanguage { get; set; }

我现在试图比较“简单”的 Linq 查询和 Rx 提供的 ToObservable() 替代方案。使用 Linq 我有以下有效的查询：

var m = new worldEntities();
        var res = m.country.Where(e => e.countrylanguage.Any(i=>i.Language.Equals("German"))).Select(e => e.Name);

但是我无法弄清楚 Rx 对应物。如果我尝试使用相同的方法

var set = m.country.ToObservable();
        set.Where(e => e.countrylanguage.Any(i => i.Language.Equals("German")))
            .Select(e => e.Name).Buffer(50).
            Subscribe(l=>
            Items.AddRange(l));

我会得到一个带有以下内部消息的异常：

{"There is already an open DataReader associated with this Connection which must be closed first."}

所以我的问题是，RX 查询应该如何实现相同的结果。

提前致谢和亲切的问候。

【问题讨论】：

标签： entity-framework linq system.reactive observable

【解决方案1】：

也许这样的事情应该有效：

    var set = m
.country
.Where(e => e.countrylanguage.Any(i => i.Language.Equals("German")))
.Select(e => e.Name)
.ToObservable();
            set.Buffer(50).
                Subscribe(l=>
                Items.AddRange(l));

然而，对于 RX 扩展来说，这是一种非常不寻常的应用程序，而且绝对不是最适合“适应”的应用程序。

【讨论】：

谢谢你，成功了。但是，您能否解释一下为什么您认为将 RX 用于数据库是一种不寻常的情况？根据link Rx 可以简化很多事情，尤其是按块填充 UI 元素。所以我从你的例子中学到的是以下内容。首先创建 linqToSQL 查询并让 EF 发挥作用，然后将结果转换为 IObservable。这应该是标准方法吗？再次感谢您的快速回复。
AFAIK Rx 事物应该基于推送数据的活动源，而您的方法并非如此。活动源可以基于定期轮询数据库以获取更改并将其推送，但不能基于对数据库的单个查询。

【解决方案2】：

数据库是基于拉取的系统。您请求从数据库中提取数据。 Rx 是关于处理推送请求，而不是拉取请求。这不是 Rx 的一个很好的用途。

如果您想学习 Rx，请开始使用它进行事件处理。

【讨论】：