如何对 Observable.Sample() 进行单元测试？

Question

我有一个类封装，Observable.Sample()例如：

class IntervalRequestScheduler
{
    private Subject<Action> _requests = new Subject<Action>();
    private IDisposable _observable;

    public IntervalRequestScheduler(TimeSpan requestLimit)
    {
        _observable = _requests.Sample(requestLimit)
                               .Subscribe(action => action());
    }

    public Task<T> ScheduleRequest<T>(Func<Task<T>> request)
    {
        var tcs = new TaskCompletionSource<T>();
        _requests.OnNext(async () =>
        {
            try
            {
                T result = await request();
                tcs.SetResult(result);
            }
            catch (Exception ex)
            {
                tcs.SetException(ex);
            }
        });
        return tcs.Task;
    }
}

我怎样才能正确测试它？我所有的尝试要么过早退出，要么导致死锁。

Answer 1

在 Rx 中控制时间

单元测试 Rx 的关键是了解如何使用TestScheduler. Rx 库中的所有基于时间的运算符都采用一个可选IScheduler参数，以便您执行此操作。您的基于时间的操作员也应该这样做。

所以我们需要做的第一件事是修改你的IntervalRequestScheduler构造函数以促进这一点：

public IntervalRequestScheduler(TimeSpan requestLimit,
                                // The scheduler is optional
                                IScheduler scheduler = null)
{
    // assign a default if necessary
    scheduler = scheduler ?? Scheduler.Default;

    // make sure to pass the scheduler in to `Sample`
    _observable = _requests.Sample(requestLimit, scheduler)
                            .Subscribe(action => action());
}

有了这个改变，我们现在可以控制时间了！

这是一个示例单元测试，它将调用IntervalRequestScheduler实例的ScheduleRequest方法十次 - 然后将时间提前一秒的示例持续时间并检查是否只有一项任务已完成：

[Test]
public void ASingleTaskIsCompletedWhenTenAreScheduledWithinInterval()
{
    var scheduler = new TestScheduler();
    var sampleDuration = TimeSpan.FromSeconds(1);

    var intervalRequestScheduler = new IntervalRequestScheduler(sampleDuration,
                                                                scheduler);

    // use a helper method to create "requests"
    var taskFactories = Enumerable.Range(0, 10).Select(CreateRequest);

    // schedule the requests and collect the tasks into an array
    var tasks =
        (from tf in taskFactories
            select intervalRequestScheduler.ScheduleRequest(tf)).ToArray();

    // prove no tasks have completed
    var completedTasksCount = tasks.Count(t => t.IsCompleted);
    Assert.AreEqual(0, completedTasksCount);

    // this is the key - we advance time simulating a sampling period.
    scheduler.AdvanceBy(sampleDuration.Ticks);

    // now we see exactly one task has completed
    completedTasksCount = tasks.Count(t => t.IsCompleted);
    Assert.AreEqual(1, completedTasksCount);
}

// helper to create requests
public Func<Task<int>> CreateRequest(int result)
{
    return () => Task.Run(() => result);
}

在旁边

到目前为止，我只关注手头的问题 - 但我确实想补充一点，实际动机IntervalRequestScheduler有点不清楚，代码看起来有点乱。在不混合包装任务和 IObservables 的情况下，可能有更好的方法来实现这一点。通过控制所涉及的调度程序，留在 Rx 世界中还可以更容易地使测试可预测。在上面的代码中，我掩盖了一些令人讨厌的地方，因为任务调用是异步的，并且在您测试它时，一个启动的任务可能实际上尚未完成 - 所以要绝对正确，您需要进入监控任务并给它们时间开始和完成的杂乱事务。但希望您能看到 TestScheduler 避免了 Rx 端的所有这些混乱。

如果您想将运行的作业数量限制在某个速率，为什么不只是对输入进行采样并投影输出呢？

例如 - 假设您处理一个Func<int,int>称为类型的请求函数runRequest和一个IObservable<int> requests输入流，提供每个请求的输入（Subject<int>例如可以是一个）。那么你可以只拥有：

requests.Sample(TimeSpan.FromSeconds(1), scheduler)
        .Select(input => request(input))
        .Subscribe(result => /* DoSomethingWithResult */);

当然不知道这是否适用于您的场景，但它可能会激发一些想法！