У меня есть наблюдаемый поток данных, к которому я применяю операции, разделяюсь на два отдельных потока, применяю более (различные) операции к каждому из двух потоков и снова объединяюсь. Я пытаюсь разделить наблюдаемое между двумя подписчиками, используяPublish а такжеConnect но каждый из подписчиков, похоже, использует отдельный поток. То есть в приведенном ниже примере я вижу «Выполнение дорогостоящей операции», напечатанное один раз для каждого элемента в потоке.для обоих подписчиков, (Представьте себе, что дорогая операция должна выполняться только один раз между всеми подписчиками, поэтому я пытаюсь повторно использовать поток.) ​​Я использовалPublish а такжеConnect попытаться поделиться объединенной наблюдаемой с обоими подписчиками, но это, кажется, имеет неправильный эффект.

Пример с вопросом:

var foregroundScheduler = new NewThreadScheduler(ts => new Thread(ts) { IsBackground = false });
var timer = Observable.Timer(TimeSpan.Zero, TimeSpan.FromSeconds(10), foregroundScheduler);
var expensive = timer.Select(i =>
{
    // Converting to strings is an expensive operation
    Console.WriteLine("Doing an expensive operation");
    return string.Format("#{0}", i);
});

var a = expensive.Where(s => int.Parse(s.Substring(1)) % 2 == 0).Select(s => new { Source = "A", Value = s });
var b = expensive.Where(s => int.Parse(s.Substring(1)) % 2 != 0).Select(s => new { Source = "B", Value = s });

var connectable = Observable.Merge(a, b).Publish();
connectable.Where(x => x.Source.Equals("A")).Subscribe(s => Console.WriteLine("Subscriber A got: {0}", s));
connectable.Where(x => x.Source.Equals("B")).Subscribe(s => Console.WriteLine("Subscriber B got: {0}", s));
connectable.Connect();

Я вижу следующий вывод:

Doing expensive operation
Doing expensive operation
Subscriber A got: { Source = A, Value = #0 }
Doing expensive operation
Doing expensive operation
Subscriber B got: { Source = B, Value = #1 }

(Вывод продолжается, сокращен для краткости.)

Как я могу поделиться наблюдаемым с обоими подписчиками?