Почему блок IObservable <T>.First()?

В последнее время я пытаюсь разобраться с Reactive Extensions для .NET, но немного нарисовал концептуальную стену: я не могу понять, почему блоки 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() до IObservable s, поэтому, вероятно, я только что сделал не то, что здесь произошло. Однако на основе различных сигнатур методов невозможно выполнить одно из следующих действий:

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

или

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

Что мне не хватает?

Ответ 1

First() возвращает a T, а не Observable<T>, поэтому он должен блокироваться.

Неблокирующая форма xs.Take(1)

Ответ 2

Попытка этого кода в тестовом проекте работала нормально, поэтому я пересмотрел проблемный код. Оказалось, проблема заключалась в том, что IObservable<ListItem> был Publish() где-то под крышками и поэтому преобразован в IConnectableObservable<ListItem>. Без вызова для подключения подписка никогда не была "активирована".