Rx 中的历史调度程序答案

【问题标题】：HistoricalScheduler in RxRx 中的历史调度程序
【发布时间】：2014-02-13 19:32:46
【问题描述】：

我有以下代码尝试重播序列；但 PostWorkItem 总是为空；我不明白为什么。 以下代码给出堆栈溢出错误。

        Task<List<WorkItem>> WorkItems = DbUtil.GetWorkItems(new List<string>() { WorkItemStatus.NEW, WorkItemStatus.PROCESSING });
         Items = WorkItems.Result;

        replay = Observable.Generate(
                Items.GetEnumerator(),
                events => events.MoveNext(),
                events => events,
                events => events.Current,
                events => System.TimeSpan.FromMilliseconds(50),
                    var);

        replay.Subscribe(
            W => 
                PostWorkItem(W)
            );

        //var.AdvanceBy(System.TimeSpan.FromSeconds(1));

        var.Start();

【问题讨论】：

标签： system.reactive reactive-programming

【解决方案1】：

您需要在Observable.Generate 中明确指定类型参数，以免与非泛型IEnumerator 混淆。尝试将第三行代码替换为：

replay = Observable.Generate<IEnumerator<WorkItem>,WorkItem>(

【讨论】：

詹姆斯谢谢；代码根据您的建议工作；但是当我使用 Scheduler.AdvanceTo(Date.Now) 时，这会产生 StackOverFlow 错误。
好吧，我可以告诉你Observable.Generate 不是递归的，所以我很确定这将取决于你使用的枚举器。我建议你设置一些断点并检查堆栈，看看里面有什么以及它是如何增长的。
感谢您的建议。通过注释掉这段代码，它不会引发堆栈溢出。 //events => events.Current.RequestedDateTime，但我想提前到 2 月 12 日。我不能这样做......我想确保 AdvanceTo 与所有项目的 RequesteDateTime 相关，所以我只能重播到 2 月 12 日，然后在下一次迭代我会提前到 2 月 14 日...