非折返在C＃

Question

鉴于以下方法观察到：

如果我离开了黑客到位，我的单元测试立即“可观察有没有数据”完成。

如果我采取了破解，有多个线程都尝试在同一时间登录。
主机服务不允许此。

如何确保只有一个线程是生产在任何给定时间点观测。

private static object obj = new object(); 
    private static bool here = true; 
    public IObservable<Party> LoadAllParties(CancellationToken token) 
    { 
     var parties = Observable.Create<Party>(
      async (observer, cancel) => 
      { 
       // this is just a hack to test behavior 
       lock (obj) 
       { 
        if (!here) 
         return; 
        here = false; 
       } 
       // end of hack. 
       try 
       { 
        if (!await this.RequestLogin(observer, cancel)) 
         return; 

        // request list. 
        await this._request.GetAsync(this._configuration.Url.RequestList); 
        if (this.IsCancelled(observer, cancel)) 
         return; 

        while (!cancel.IsCancellationRequested) 
        { 
         var entities = await this._request.GetAsync(this._configuration.Url.ProcessList); 
         if (this.IsCancelled(observer, cancel)) 
          return; 

         var tranche = this.ExtractParties(entities); 

         // break out if it's the last page. 
         if (!tranche.Any()) 
          break; 

         Array.ForEach(tranche, observer.OnNext); 

         await this._request.GetAsync(this._configuration.Url.ProceedList); 
         if (this.IsCancelled(observer, cancel)) 
          return; 
        } 

        observer.OnCompleted(); 
       } 
       catch (Exception ex) 
       { 
        observer.OnError(ex); 
       } 
      }); 
     return parties; 
    } 

我的单元测试：

var sut = container.Resolve<SyncDataManager>(); 
var count = 0; 
var token = new CancellationTokenSource(); 
var observable = sut.LoadAllParties(token.Token); 
observable.Subscribe(party => count++); 
await observable.ToTask(token.Token); 
count.Should().BeGreaterThan(0); 

Answer 1

我认为你的问题是从XY Problem痛苦 - 代码中包含几个电话不包括可能包含重要的副作用的方法，我觉得那回事可用的信息不会导致最好的建议。

这就是说，我怀疑你没有打算订阅observable两次 - 一次用明确的Subscribe打电话，一次用ToTask()打电话。这肯定会解释并发呼叫，这是发生在两个不同的订阅。

编辑：

如何断言的长度，而不是（调整以适应超时）：

var length = await observable.Count().Timeout(TimeSpan.FromSeconds(3)); 

更好的方式是寻找到RX-测试和嘲笑你的依赖。这是一个很大的话题，但this long blog post from the Rx team explains it very well和关于TPL-Rx相互作用的这个答案可能有所帮助：Executing TPL code in a reactive pipeline and controlling execution via test scheduler