测试 Rx 和 TcpClient

Question

经过相当多的思考和在互联网上翻找，我想出了（我认为是）将 TcpIp 转换为 Observable 数据流的非常漂亮的封装：

public static IObservable<string> CreateFromTcpIp(IPAddress ip, int port)
{
    return 
        Observable.Using(
            () => new TcpClient(),                               //create client
            client => client.ConnectObservable(ip, port)         //connect async
                      .Select(_ => client.GetStream())            // get stream
                      .Select(stream => new BinaryReader(stream)) // get reader 
                      .SelectMany(reader => reader.ToObservable()) // see below
                      .Retry()                                // if error, retry 
            )
            .Publish().RefCount(); //only subscribe once for all subscribers
}

其中ConnectObservable 就是client.ConnectAsync(host, port).ToObservable()，而reader.ToObservable 看起来像这样：

private static IObservable<string> ToObservable(this BinaryReader source)
{
   return Observable.Create<string>(
       observer =>
       {
           return Observable.Generate(
               source,
               reader => true, //just keep on going
               reader => reader,
               reader => reader.ReadString())
           .Subscribe(observer);
       });
}

我面临的问题是如何进行测试。我不想在调用堆栈中一直实现/包装接口。

我一直在尝试在我的测试类中使用 TcpListener 创建一个“服务器”，但是遇到了各种（可能不相关的）问题，让订阅和服务共存（订阅我的 CreateFromTcpIp 的结果似乎永远不会返回 IDisposable，在我的测试中插入延迟似乎最终会锁定程序集，因此我无法重建）。

是否有我遗漏的替代方法？

EDIT 订阅不返回的问题与对有故障的 Observable 进行重试有关。我问了另一个问题here。

Answer 1

我想你可以像这样测试BinaryReaderToObservable（如果你让它可以访问的话）。尽管我编写它是为了传递所写的代码（使用 nunit），但它期望异常的事实可能会突出一些问题：

[Test]
public void TestStreamToObservable()
{

    var expectedText = new List<string>
    {
        "A good test is simple.",
        "A rolling stone gathers no moss.",
        "Test properly"
    };

    var stream = new MemoryStream();
    var writer = new BinaryWriter(stream);
    expectedText.ForEach(writer.Write);
    writer.Flush();

    stream.Seek(0, SeekOrigin.Begin);
    var reader = new BinaryReader(stream);

    var resultText = new List<string>();

    Assert.Throws<EndOfStreamException>(
        () => reader.ToObservable().Subscribe(resultText.Add));

    CollectionAssert.AreEquivalent(expectedText, resultText);
}

至于其余部分，您通过创建一个new TcpClient()，除了集成测试之外，您已经把自己搞砸了。 new 的任何具体的东西总是会给测试带来麻烦。您可以重构并将Retry().Publish().RefCount() 拉出到一个单独的方法中，该方法在IObservable<string> 上运行。然后将TcpClient 位仅留给集成测试。

Answer 2

好的，为了后代，这是我在解决这个问题时学到的一些东西。 主要是关于使用 Rx 的微妙之处：

• 如果你没有观察到OnError，它会抛出异常。
• 除此之外，如果您在 Task 上执行了 ToObservable 并且您没有观察到 OnError，它只会在您取消订阅时抛出
• 此外，在任务下游执行Retry 毫无意义：一旦引发异常，即为Result，它不会再尝试（我花了一段时间才解决这个问题！）。
• 如果我的订阅永远不会返回，我需要 SubscribeOn 另一个线程 (Scheduler.Default)。
• James 的回答中暗示了另一个错误：BinaryReader 上的 ToObservable 引发异常。这里重要的是它是Exception，而不是OnError。所以不可能有Retry。

（小代码sn-p向我们自己证明这一点）：

Observable.Generate(0, i => i < 4, i => i + 1, 
                    i => { if(i > 2) throw new Exception("sorry"); return i; })
          .Subscribe(i => Console.WriteLine(i), 
                     ex => Console.WriteLine(ex.Message));

鉴于此，我重新编码了我的ToObservable 以使用它（不太漂亮）：

return Observable.Create<string>(obs =>
{
    var subscribed = true;
    var errored = false;
    string result = String.Empty;
    while(subscribed && !errored)
    {
        try
        {
            result = source.ReadString();
        }
        catch(Exception ex)
        {
            errored = true;
            obs.OnError(ex);
        }
        obs.OnNext(result); 
    }
    return Disposable.Create(() => subscribed = false);
});

• 为了进一步发展，我想要一个不那么激进的Retry。这让我找到了this excellent answer。 （似乎不管我在 Rx 中写了什么，总有一种更优雅的方式来做到这一点）
• James 的 Spy 扩展方法非常很有帮助。强烈推荐。