Почему IObservable <T>.First () блокируется?

Question

В последнее время я пытался разобраться с 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 с, так что, скорее всего, я тут что-то не так сделал.Тем не менее, это невозможно, основываясь на различных сигнатурах методов:

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)