将生成的事件序列创建为冷序列

Question

FWIW - 在asking for advice on meta

我有一个包含配置数据的网络服务。我想定期调用它Tok，以刷新使用它的应用程序中的配置数据。如果服务出错（超时、停机等），我想保留上一次调用的数据，并在不同的时间间隔 Tnotok 后再次调用服务。最后，我希望行为是可测试的。

由于管理时间序列和可测试性似乎是响应式扩展的一个强项，我开始使用将由生成的序列提供的 Observable。以下是我创建序列的方式：

Observable.Generate<DataProviderResult, DataProviderResult>(
    // we start with some empty data
    new DataProviderResult() { 
            Failures = 0
            , Informations = new List<Information>()},
    // never stop
    (r) => true,
    // there is no iteration
    (r) => r,
    // we get the next value from a call to the webservice
    (r) => FetchNextResults(r),
    // we select time for next msg depending on the current failures
    (r) => r.Failures > 0 ? tnotok : tok,
    // we pass a TestScheduler
    scheduler)
.Suscribe(r => HandleResults(r));

我目前有两个问题：

---

看起来我正在创建一个热门的 observable。即使尝试使用Publish/Connect，我的订阅操作也缺少第一个事件。如何将其创建为冷可观察对象？

myObservable = myObservable.Publish();
myObservable.Suscribe(r => HandleResults(r));
myObservable.Connect() // doesn't call onNext for first element in sequence

---

当我订阅时，订阅和生成的顺序似乎不正确，因为对于任何帧，订阅方法都会在 FetchNextResults 方法之前触发。正常吗？我希望序列调用帧 f 的方法，而不是 f+1。 这是我用于获取和订阅的代码：

private DataProviderResult FetchNextResults(DataProviderResult previousResult)
{
    Console.WriteLine(string.Format("Fetching at {0:hh:mm:ss:fff}", scheduler.Now));
    try
    {
        return new DataProviderResult() { Informations = dataProvider.GetInformation().ToList(), Failures = 0};
    }
    catch (Exception)
    {}
    previousResult.Failures++;

    return previousResult;
}

private void HandleResults(DataProviderResult result)
{
    Console.WriteLine(string.Format("Managing at {0:hh:mm:ss:fff}", scheduler.Now));
    dataResult = result;
}

以下是我所看到的，促使我提出这些问题：

Starting at 12:00:00:000
Fetching at 12:00:00:000 < no managing the result that has been fetched here
Managing at 12:00:01:000 < managing before fetching for frame f
Fetching at 12:00:01:000
Managing at 12:00:02:000
Fetching at 12:00:02:000

---

编辑：这是一个简单的可复制粘贴程序，可以说明问题。

/*using System;
using System.Reactive.Concurrency;
using System.Reactive.Linq;
using Microsoft.Reactive.Testing;*/

private static int fetchData(int i, IScheduler scheduler)
{
  writeTime("fetching " + (i+1).ToString(), scheduler);
  return i+1;
}
private static void manageData(int i, IScheduler scheduler)
{
  writeTime("managing " + i.ToString(), scheduler);
}
private static void writeTime(string msg, IScheduler scheduler)
{
  Console.WriteLine(string.Format("{0:mm:ss:fff} {1}", scheduler.Now, msg));
}

private static void Main(string[] args)
{
    var scheduler = new TestScheduler();
    writeTime("start", scheduler);
    var datas = Observable.Generate<int, int>(fetchData(0, scheduler),
                                                (d) => true,
                                                (d) => fetchData(d, scheduler),
                                                 (d) => d,
                                                 (d) => TimeSpan.FromMilliseconds(1000),
                                                 scheduler)
                                                 .Subscribe(i => manageData(i, scheduler));

    scheduler.AdvanceBy(TimeSpan.FromMilliseconds(3000).Ticks);
}

这会输出以下内容：

00:00:000 start
00:00:000 fetching 1
00:01:000 managing 1
00:01:000 fetching 2
00:02:000 managing 2
00:02:000 fetching 3

我不明白为什么在获取第一个元素后没有立即对其进行管理。在有效拉取数据和将数据交给观察者的序列之间有一秒钟的时间。我在这里遗漏了什么还是预期的行为？如果是这样，有没有办法让观察者立即对新值做出反应？

Answer 1

您误解了timeSelector 参数的用途。每次生成一个值时都会调用它，并返回一个时间，该时间指示在将该值传递给观察者之前延迟多长时间，然后生成下一个值。

这是一种解决问题的非生成方法。

private DataProviderResult FetchNextResult()
{
    // let exceptions throw
    return dataProvider.GetInformation().ToList();
}

private IObservable<DataProviderResult> CreateObservable(IScheduler scheduler)
{
    // an observable that produces a single result then completes
    var fetch = Observable.Defer(
        () => Observable.Return(FetchNextResult));

    // concatenate this observable with one that will pause
    // for "tok" time before completing.
    // This observable will send the result
    // then pause before completing.
    var fetchThenPause = fetch.Concat(Observable
        .Empty<DataProviderResult>()
        .Delay(tok, scheduler));

    // Now, if fetchThenPause fails, we want to consume/ignore the exception
    // and then pause for tnotok time before completing with no results
    var fetchPauseOnErrors = fetchThenPause.Catch(Observable
        .Empty<DataProviderResult>()
        .Delay(tnotok, scheduler));

    // Now, whenever our observable completes (after its pause), start it again.
    var fetchLoop = fetchPauseOnErrors.Repeat();

    // Now use Publish(initialValue) so that we remember the most recent value
    var fetchLoopWithMemory = fetchLoop.Publish(null);

    // YMMV from here on.  Lets use RefCount() to start the
    // connection the first time someone subscribes
    var fetchLoopAuto = fetchLoopWithMemory.RefCount();

    // And lets filter out that first null that will arrive before
    // we ever get the first result from the data provider
    return fetchLoopAuto.Where(t => t != null);
}

public MyClass()
{
    Information = CreateObservable();
}

public IObservable<DataProviderResult> Information { get; private set; }

Answer 2

Generate 产生冷的可观察序列，所以这是我的第一个警钟。

我尝试将您的代码拉入 linqpad* 并运行它并对其进行了一些更改以专注于问题。在我看来，您混淆了 Iterator 和 ResultSelector 函数。这些是从后到前的。当您迭代时，您应该从上次迭代中获取值并使用它来生成下一个值。结果选择器用于从您正在迭代的实例中挑选（选择）值。

所以在你的情况下，你迭代的类型是你想要产生值的类型。因此，请将您的 ResultSelector 函数保留为标识函数x=>x，并且您的 IteratorFunction 应该是进行 WebService 调用的那个。

Observable.Generate<DataProviderResult, DataProviderResult>(
    // we start with some empty data
    new DataProviderResult() { 
            Failures = 0
            , Informations = new List<Information>()},
    // never stop
    (r) => true,
    // we get the next value(iterate) by making a call to the webservice
    (r) => FetchNextResults(r),
    // there is no projection
    (r) => r,
    // we select time for next msg depending on the current failures
    (r) => r.Failures > 0 ? tnotok : tok,
    // we pass a TestScheduler
    scheduler)
.Suscribe(r => HandleResults(r));

作为旁注，请尝试在迭代时更喜欢不可变类型而不是改变值。

*请提供代码的自主工作 sn-p，以便人们可以更好地回答您的问题。 :-)