c# - 如何将 TestScheduler 与 ReplaySubject 时间窗一起使用

Question

我有一个测试代码的例子:

    [Fact]
    public void Should_only_contain_most_recent() {
        var window = TimeSpan.FromMilliseconds(200);
        var results = new ReplaySubject<long>(window);
        results.OnNext(1);
        results.OnNext(2);
        System.Threading.Thread.Sleep(50);
        results.OnNext(3);
        System.Threading.Thread.Sleep(50);
        results.OnNext(4);
        System.Threading.Thread.Sleep(200);
        results.OnNext(5);
        results.OnCompleted();
        var items = results.ToEnumerable();
        Assert.True(items.SequenceEqual(new long [] { 5 }));
    }

我想把它变成使用 Microsoft.Reactive.Testing 包中的 TestScheduler，因为我想摆脱这里的 sleep 。

我已经尝试过这样做了。

    [Fact]
    public void Should_only_contain_most_recent() {
        var scheduler = new TestScheduler();
        var window = TimeSpan.FromMilliseconds(200);
        var results = new ReplaySubject<long>(window /* Never finishes when add the scheduler here.*/);
        results.OnNext(1);
        results.OnNext(2);
        scheduler.Schedule(TimeSpan.FromMilliseconds(50), () => results.OnNext(3));
        scheduler.Schedule(TimeSpan.FromMilliseconds(50), () => results.OnNext(4));
        scheduler.Schedule(TimeSpan.FromMilliseconds(200), () => results.OnNext(5));
        scheduler.Schedule(TimeSpan.FromMilliseconds(210), results.OnCompleted);
        scheduler.Start();

        var items = results.ToEnumerable();
        Assert.True(items.SequenceEqual(new long[] { 5 }));
    }

但是我错过了一些东西。

Answer 1

推测您正在尝试测试 ReplaySubject 的重放行为。问题是您需要启动 TestScheduler 两次。它只运行到所有计划的事件都已执行 - 但我们需要在没有任何订阅者的情况下加载 ReplaySubject 以便它具有将重播的事件 - 然后在订阅者连接后再次运行调度程序。

这是编写此测试的惯用方法:

首先，我用我们需要的事件设置source。通过从 ReactiveTest 派生测试类，您可以利用 OnXXX 辅助方法更轻松地创建源事件。

准备好源流后，我们可以订阅 ReplaySubject 并在 TestScheduler 上调用 Start。这将运行虚拟时间直到最后一个预定事件 (T=300ms) - 我通过转储超时来演示。

现在我们使用 TestScheduler 的 CreateObserver 方法为结果创建记录器并将其订阅到 ReplaySubject。

现在我们再次启动 TestScheduler - 这会从 T=300ms 开始计时，直到刷新新生成的计划事件。这些是 ReplaySubject 需要重播的事件 - 那些在 200 毫秒窗口内的事件。请注意，这些将在 T=300 毫秒 + 1 个滴答、+ 2 个滴答和 + 3 个滴答时发出 - 这是因为 TestScheduler 具有 1 个滴答的分辨率。

代码如下:

public class MyTests : ReactiveTest
{
    [Fact]
    public void Should_only_contain_most_recent()
    {
        var scheduler = new TestScheduler();

        var source = scheduler.CreateHotObservable(
            OnNext(0, 1),
            OnNext(0, 2),
            OnNext(TimeSpan.FromMilliseconds(50).Ticks, 3),
            OnNext(TimeSpan.FromMilliseconds(100).Ticks, 4),
            OnNext(TimeSpan.FromMilliseconds(300).Ticks, 5),
            OnCompleted(TimeSpan.FromMilliseconds(300).Ticks, witness: 0));

        var replaySubject = new ReplaySubject<int>(
            TimeSpan.FromMilliseconds(200), scheduler);

        source.Subscribe(replaySubject);

        scheduler.Start();

        /* the test scheduler is now at 300 milliseconds
            * and the ReplaySubject is loaded */
        Console.WriteLine(scheduler.Now.Ticks);

        var results = scheduler.CreateObserver<int>();
        replaySubject.Subscribe(results);

        /* run the scheduler on to flush the events from the ReplaySubject */
        scheduler.Start();

        results.Messages.AssertEqual(
            OnNext(TimeSpan.FromMilliseconds(300).Ticks + 1, 4),
            OnNext(TimeSpan.FromMilliseconds(300).Ticks + 2, 5),
            OnCompleted(TimeSpan.FromMilliseconds(300).Ticks + 3, witness: 0));
    }
}