非重放热可观察

Question

原始问题

我有一个场景，我有多个 IObservable 序列，我想将它们与 Merge 组合然后收听。但是，如果其中一个产生错误，我不希望它使其他流的所有内容崩溃，也不想重新订阅序列（这是一个“永久”序列）。

我通过在合并之前将Retry() 附加到流中来做到这一点，即：

IEnumerable<IObservable<int>> observables = GetObservables();

observables
    .Select(o => o.Retry())
    .Merge()
    .Subscribe(/* Do subscription stuff */);

但是，当我想对此进行测试时，问题就出现了。我想测试的是，如果observables 中的IObservable 之一产生OnError，其他的应该仍然能够发送它们的值并且它们应该得到处理

我以为我会在observables 中使用两个Subject<int>s 代表两个IObservables；一个发送OnError(new Exception())，另一个发送OnNext(1)。但是，Subject<int> 似乎会为新订阅重放所有以前的值（实际上是Retry()），从而将测试变成无限循环。

我试图通过创建手册IObservable 来解决它，该手册在第一次订阅时会产生错误，然后是空序列，但感觉很hacky：

var i = 0;
var nErrors = 2;
var testErrorObservableWithOneErrorAndThenCompletion = Observable.Create<int>(o => {
    i++;
    if (i < nErrors) {
        return Observable.Throw<int>(new Exception()).Subscribe(o);
    } else {
        return Observable.Empty<int>().Subscribe(o);
    }
});

我是否以错误的方式使用Subject 或考虑Retry()？对此还有其他想法吗？您将如何解决这种情况？

更新

好的，这是我想要的和 想 Retry() 的大理石图。

o = message, X = error.
------o---o---X
               \
     Retry() -> \---o---o---X
                             \
                   Retry() -> \...

我的问题可能更多是因为我没有一个好的股票类来使用前测试，因为Subject 想要重放我以前的所有错误。

更新 2

这是一个测试用例，显示了我对 Subject 重放其值的意思。如果我说它以 cold 的方式这样做，我是否正确使用了该术语？我知道Subject 是一种创建热可观察对象的方式，但这种行为对我来说仍然感觉“冷”。

var onNext = false;
var subject = new Subject<int>();

subject.Retry().Subscribe(x => onNext = true);
subject.OnError(new Exception());
subject.OnNext(1);

Assert.That(onNext, Is.True);

Answer 1

根据您更新的要求（您希望重试失败的 observable，而不是只想忽略它们），我们可以提出一个可行的解决方案。

首先，了解冷可观察对象（在每个订阅上重新创建）和热可观察对象（无论订阅如何都存在）之间的区别很重要。你不能 Retry() 一个热门的 observable，因为它不知道如何重新创建底层事件。也就是说，如果一个 hot observable 错误，它就永远消失了。

Subject 创建一个 hot observable，从某种意义上说，您可以在没有订阅者的情况下调用 OnNext，它会按预期运行。要将热 observable 转换为冷 observable，您可以使用 Observable.Defer，它将包含该 observable 的“订阅创建”逻辑。

说了这么多，下面是修改后的原始代码：

var success = new Subject<int>();
var error = new Subject<int>();

var observables = new List<IObservable<int>> { Observable.Defer(() => {success = new Subject<int>(); return success.AsObservable();}), 
                                               Observable.Defer(() => {error = new Subject<int>(); return error.AsObservable();}) };                                            

observables
.Select(o => o.Retry())
.Merge()
.Subscribe(Console.WriteLine, Console.WriteLine, () => Console.WriteLine("done"));

以及测试（和之前类似）：

success.OnNext(1);
error.OnError(new Exception("test"));
success.OnNext(2);
error.OnNext(-1);
success.OnCompleted();
error.OnCompleted();

和预期的输出：

1
2
-1
done

当然，您需要根据您的底层 observable 对这个概念进行重大修改。使用主题进行测试与实际使用它们不同。

我还想注意这条评论：

但是，Subject 似乎会重播所有以前的值 新订阅（实际上是 Retry() ），将测试变成 一个无限循环。

不是真的 - Subject 不是这样的。基于Retry 重新创建订阅的事实，您的代码的其他一些方面会导致无限循环，并且订阅会在某些时候产生错误。

---

原始答案（用于补全）

问题是Retry() 没有做你想让它做的事情。从这里开始：

http://msdn.microsoft.com/en-us/library/ff708141(v=vs.92).aspx

重复源 observable 序列 retryCount 次或直到 它成功终止。

这意味着Retry 将不断尝试并重新连接到底层的可观察对象，直到它成功并且不抛出错误。

我的理解是，您实际上希望观察对象中的异常忽略，而不是重试。这将做你想做的事：

observables
.Select(o => o.Catch((Func<Exception,IObservable<int>>)(e => Observable.Empty<int>())))
.Merge()
.Subscribe(/* subscription code */);

这使用Catch 捕获带有异常的可观察对象，并在此时将其替换为空的可观察对象。

这是一个使用科目的完整测试：

var success = new Subject<int>();
var error = new Subject<int>();

var observables = new List<IObservable<int>> { success.AsObservable(), error.AsObservable() };

observables
.Select(o => o.Catch((Func<Exception,IObservable<int>>)(e => Observable.Empty<int>())))
.Merge()
.Subscribe(Observer.Create<int>(Console.WriteLine, Console.WriteLine, () => Console.WriteLine("done")));

success.OnNext(1);
error.OnError(new Exception("test"));
success.OnNext(2);
success.OnCompleted();

正如预期的那样，这会产生：

1
2
done