Eu tenho um fluxo de dados observável ao qual estou aplicando operações, dividindo em dois fluxos separados, aplicando mais operações (distintas) a cada um dos dois fluxos e mesclando novamente. Estou tentando compartilhar o observável entre dois assinantes usandoPublish eConnect mas cada um dos assinantes parece estar usando um fluxo separado. Ou seja, no exemplo abaixo, vejo "Executando uma operação cara" impressa uma vez para cada item no fluxopara ambos os assinantes. (Imagine a operação cara como algo que deveria acontecer apenas uma vez entre todos os assinantes; portanto, estou tentando reutilizar o fluxo.) Eu useiPublish eConnect para tentar compartilhar o observável mesclado com os dois assinantes, mas parece ter o efeito errado.

Exemplo com o problema:

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();

Eu vejo a seguinte saída:

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 }

(A saída continua, truncada por questões de brevidade.)

Como posso compartilhar o observável com os dois assinantes?