为什么IObservable <T> .First()阻止？

Question

我一直在努力试探最近的.NET Reactive Extensions,但是已经触及了一些概念墙:我无法弄清楚为什么IObservable.First()会阻塞.

我有一些示例代码看起来像这样:

var a = new ListItem(a);
var b = new ListItem(b);
var c = new ListItem(c);
var d = new ListItem(d);

var observableList = new List<ListItem> { a,b,c,d }.ToObservable();

var itemA = observableList.First();

// Never reached
Assert.AreEqual(expectedFoo, itemA.Foo);

我期待发生的事情是itemA在引用上等于a并且能够访问其成员等.相反,发生的是First()块和Assert.AreEqual()从未到达的块.

现在,我知道在使用Rx时,代码应该Subscribe()是IObservables,所以很可能我在这里做错了.但是,根据各种方法签名,不可能执行以下任一操作:

observableList.First().Subscribe(item => Assert.AreEqual(expectedFoo, item));

要么

observableList.Subscribe(SomeMethod).First() // This doesn't even make sense, right?

我错过了什么？

Answer 1

在测试项目中尝试使用此代码工作正常,因此我重新查看了有问题的代码.事实证明问题是因为它IObservable<ListItem>被Publish()封装在某个地方,因此被转换成了一个IConnectableObservable<ListItem>.没有连接调用,订阅从未"激活".

Answer 2

First()返回一个T,而不是一个Observable<T>,所以它必须阻止.

非阻塞形式是 xs.Take(1)