IObservable等待获取所有元素错误

我有这个课：

public class TestService
{
     public IObservable<int> GetObservable(int max)
     {
         var subject = new Subject<int>();
         Task.Factory.StartNew(() =>
                               {
                                   for (int i = 0; i < max; i++)
                                   {
                                       subject.OnNext(i);
                                   }
                                   subject.OnCompleted();

                               });
         return subject;
     }
}

我也为此编写了一个测试方法：

[TestMethod]
public void TestServiceTest1()
{
   var testService = new TestService();
   var i = 0;
   var observable = testService.GetObservable(3);
   observable.Subscribe(_ =>
   {
      i++;
   });          
   observable.Wait();
   Assert.AreEqual(i, 3);
}

但是有时候我会收到错误消息：Sequence方法Wait()中不包含任何元素.

我建议在测试达到observable.Wait()行之前完成我的IObservable.
如何避免此错误？

解决方法:

在我看来,这段代码中的基本问题是IObservable代表如何观察事物的契约.

在这种情况下,GetObservable方法不仅返回合同,而且还立即执行工作(使用TPL).如果您认为可以多次订阅同一IObservable实例(这实际上是在示例代码中发生的,因为您是第一次使用Subscribe订阅,而另一个使用Wait订阅),则这没有任何意义.这个单一的区别是我学习Rx的最大绊脚石.

我将改为实现这样的方法(并避免完全使用Subject<&gt ;,因为它不是创建Observable的首选方式)：

public class TestService
{
     public IObservable<int> GetObservable(int max)
     {
         return Observable.Create<int>((IObserver<int> observer) =>
                               {
                                   for (int i = 0; i < max; i++)
                                   {
                                       observer.OnNext(i);
                                   }
                                   observer.OnCompleted();
                               });
     }
}

Observable.Create也有一些有趣的替代方法,它们提供了更大的灵活性.

标签：system-reactive,reactive-programming,c
来源： https://codeday.me/bug/20191031/1972068.html