Невоспроизведение горячего наблюдаемого

Исходный вопрос

У меня есть сценарий, в котором у меня есть несколько последовательностей IObservable, которые я хочу объединить с Слитьсяи потом слушать. Однако, если один из них выдает ошибку, я не хочу, чтобы он разрушил все для других потоков, а также повторно подписался на последовательность (это «вечная» последовательность).

Я делаю это, добавляя Retry()к потокам перед слиянием, то есть:

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

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

Однако проблема возникает, когда я хочу проверить это. Что я хотел бы проверить, так это то, что если один из IObservableв observablesвыдает OnError, другие должны по-прежнему иметь возможность отправлять свои значения через и они должны быть обработаны

Я думал, что просто использовал бы два Subject, представляющих два IObservableв observables; один отправляет OnError(new Exception()), а другой после этого отправляет OnNext(1). Однако похоже, что Subjectвоспроизведет все предыдущие значения для новой подписки (фактически Retry()так и есть), превратив тест в бесконечный цикл.

Я попытался решить эту проблему, создав руководство IObservable, которое выдает ошибку при первой подписке, а затем пустую последовательность, но это кажется хакерским:

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() -> \...

Моя проблема, возможно, больше в том, что у меня нет хорошего класса акций для использования предварительного тестирования, так как Субъектхочет воспроизвести все мои предыдущие ошибки.

Обновление 2

Вот тестовый пример, который показывает, что я имею в виду, говоря о Субъекте, воспроизводящем свои значения. Правильно ли я использую этот термин, если говорю, что он делает это холодноспособом? Я знаю, что 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);