热Observable经Select派生后执行Take(1) await不再推送值的原因求解

问题原因 #

这个问题的根源是Publish().RefCount()的工作机制：

• RefCount会维护当前的订阅者计数，当第一个订阅者出现时自动连接上游Observable，当最后一个订阅者取消订阅时，自动断开上游连接，停止序列推送。
• 你调用await o2.Take(1)时，Take(1)会生成一个临时订阅：
  1. 临时订阅触发RefCount计数变为1，上游Interval开始运行，1秒后推送第一个值
  2. 拿到值后Take(1)会立即主动取消这个临时订阅，此时RefCount计数从1降到0，直接触发上游连接断开，o1就停止推送了。

你后续再订阅o2的时候，RefCount会重新连接上游Interval，需要再等1秒才会推送新值，如果你没有等待足够时间就操作，就会误以为序列不再工作。

加Task.Delay(1)后问题消失的原因 #

这是竞态条件导致的巧合：
Take(1)的订阅取消操作在Rx的默认调度器上是异步排队执行的，如果你在await Take(1)之后立刻订阅o2，此时订阅取消还没执行完成，RefCount计数还是1，新订阅进来后计数变成2，等订阅取消执行完成后计数回到1，就不会触发上游断开，序列会继续运行。
如果你在await之后加了Console.ReadLine()，等待用户输入的时间足够让订阅取消执行完成，RefCount计数降到0，上游断开。而你加Task.Delay(1)的测试场景下，大概率是你等待Delay完成后立刻按回车订阅，刚好卡在上游还没完全断开的时间窗口，所以序列能继续运行，这个效果是不稳定的。

正确实现方案 #

如果你需要o1是持续运行的热Observable，不会因为暂时没有订阅者就停止，可以把RefCount换成手动调用Connect()：

int i = 1;
var o1 = Observable
    .Interval(TimeSpan.FromSeconds(1))
    .Select(x => { i++;return i; })
    .Do(x => Console.WriteLine($"emit {x}"))
    .Publish();
// 手动启动上游，不会因为没有订阅者停止
o1.Connect();

var o2 = o1.Select(x => x + 5);

await o2.Take(1);
// 后续订阅会直接接收当前正在推送的序列值，不需要重新启动
using (o2.Subscribe(x =>
 {
     Console.WriteLine($"output {x}");
 }))
{
    Console.WriteLine($"subscrbied");
    Console.ReadLine();
}
Console.ReadLine();

内容的提问来源于stack exchange，提问作者user5182794