围绕响应式扩展的大多数示例代码都围绕着如何在序列上组合逻辑和运算符。
围绕 Observable 生成的部分主要围绕“FromEventPatter”、“FromAsynch”等。
IObservable<string> observableHotStatus = ??;
foreach (var task in todo)
{
//Process task;
//Post status message into observable; How do I do this?
}
简而言之，我想要一个可以发布到的对象，例如 ActionBlock、Action (of T) 或类似的东西。 实现这一目标的最简单方法是什么？
【问题讨论】：
编辑：
更仔细地检查您的代码，我建议使用Observable.Create。即使它只返回一个 cold observable，您也可以将 Publish 运算符应用于生成的 observable 以使其 hot。
如果task 你实际上指的是Task<T>，那么你可以使用Observable.Create 的重载，它允许你定义一个异步迭代器。例如：
IObservable<string> statuses = Observable.Create<string>(
(observer, cancel) =>
{
foreach (var task in todo)
{
cancel.ThrowIfCancellationRequested();
await task;
observer.OnNext("Status");
}
});
上一个答案：
您可以使用以下类型之一，但我建议您在做出决定之前先阅读To Use Subject or Not To Use Subject。
Subject<T>：通用，类似“事件”，hot observable。调用 OnNext 就像发起一个经典的 .NET 事件。BehaviorSubject<T>：一般用作属性的后备字段，它表示可观察到的变化“事件”序列。每当观察者订阅时，它会立即收到当前值，然后是对属性的所有更改。您可以随时从Value 属性中提取当前值；例如，在您的财产的吸气剂中。在您的属性的设置器中调用OnNext，您不必保留重复的支持字段。它也是 Rx 的 continuous function 版本，如果我对 FRP 的理解是正确的，它是您在 Rx 中可以找到的唯一类似 FRP 的东西。ReplaySubject<T>：通常用作“事件”的历史缓冲区，它表示一个可观察的值序列，从观察者错过的值开始，每当观察者订阅时。您可以控制缓冲多远的值；它就像价值历史的滑动窗口。您很少需要使用这种类型。在大多数情况下，Observable.Replay 运算符就可以了。AsyncSubject<T>：一般用来捕获hot的结果，像Task<T>这样的异步函数。您很少需要使用这种类型。在大多数情况下，Observable.FromAsyncPattern 或 Task-conversion 运算符都可以。【讨论】：