Why does IObservable<T>.First() block?

Question

I've been trying to get my head around the Reactive Extensions for .NET of late, but have hit a bit of a conceptual wall: I can't work out why IObservable.First() blocks.

I have some sample code that looks a bit like this:

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);

What I was expecting to happen was for itemA to be referentially equal to a and to be able to access its members, etc. What happens instead is that First() blocks and the Assert.AreEqual() is never reached.

Now, I know enough to know that when using Rx, code should Subscribe() to IObservables, so it's likely that I've just done the wrong thing here. However, it's not possible, based on the various method signatures to do either of:

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

or

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

What am I missing?

Answer 1

Trying this code out in a test project worked fine, so I revisited the problematic code. It turned out the problem was because the IObservable<ListItem> was being Publish()ed somewhere under the covers and so being converted into an IConnectableObservable<ListItem>. Without a call to connect, the subscription was never "activated".

Answer 2

First() returns a T, not an Observable<T>, so it must block.

A non-blocking form is xs.Take(1)