使用Rx Repeat()和Replay()缓存并重新启动DNS查询

Question

我是一个Rx新手,所以我希望你能忍受我.作为我自己的练习,也可能是我可以为同事演示的样本,我为Dns.BeginGetHostEntry()/ EndGetHostEntry()完成了两个包装类:DnsResolver和DnsResolverRx.

每个类都有一个公共静态方法:

void Resolve(string host, Action<IPHostEntry> getResult, Control context = null);

...以及使其变得有趣的一些其他要求:1.如果提供了上下文,则必须在关联的线程2上调用getResult.对于MaxResultAge秒,将缓存同一主机的先前结果.

非Rx版本工作正常,但与此问题并不真正相关.Rx版本如下所示:

class DnsResolverRx
{
  static Func<string, IObservable<IPHostEntry>> _resolver = Observable.FromAsyncPattern<string, IPHostEntry>(Dns.BeginGetHostEntry, Dns.EndGetHostEntry);

  public static void Resolve(string host, Action<IPHostEntry> setResult, Control context = null)
  {
    IObservable<IPHostEntry> result;
    result = _cache.GetOrCreateValue( // a trivial TryGetValue wrapper
      host,
      () => _resolver(host)
        .Do(e => Debug.WriteLine("resolved"))
        .Repeat()
        .Do(e => Debug.WriteLine("repeated"))
        .Replay(MaxResultAge)
        .RefCount()
    );

    result = result.Take(1); // each request needs only 1 result

    if (context != null)
      result = result.ObserveOn(context);

    result.Subscribe(
      entry => setResult(entry),
      ex => setResult(null)
    );
  }
}

static void Main(string[] args)
{
  for (int i=0; i<10; ++i)
  {
    int num = i;
    Debug.WriteLine("start" + num);
    DnsResolverRx.Resolve("chief", e => Debug.WriteLine("done"+num));
    Thread.Sleep(200);
  }

  Console.ReadLine();
}

Replay()似乎有效,因此MaxResultAge中的第一个请求都已完成,重用相同的结果.但是,下一个请求会触发Repeat(),最终会出现一个看似无穷无尽的循环:

start0
start1
start2
start3
start4
start5
resolved
repeated
done0
done1
done2
done3
done4
done5
start6
resolved
repeated
resolved
repeated
... and so on ad infinitum

任何人都可以告诉我发生了什么,以及我做错了什么？

Answer 1

好吧 - 刚刚实现FromAsyncPatterns是一次性的,基本上不能重新启动,并且它们在引用时开始,而不是在订阅时开始.

更换

_resolver(host)

同

Observable.Defer(() => _resolver(host))

似乎可以做到这一点.

如果没有一个对象(我没有发现其他问题),我很快就会接受这个答案.