对事件的操作

Question

我有一个Action<Action> operation，它执行一些长时间运行的工作，并通过提供的Action参数发送检测信号。在调用它之前，我想设置一个IObservable<Unit>，每次操作发送心跳时，它都会生成Next元素。我该怎么做

创建IObservable并没有那么复杂，但我猜概念上最简单的方法是从事件中创建它，但遗憾的是，您无法在方法体中创建事件（除非您知道方法）。我可以使用Observable.FromEvent(Action<Action> addHandler, Action<Action> removeHandler)，但这将需要丑陋的封闭操作来将方法调用转发到（参见下面的示例）
还有更优雅的方式吗？

Action forward = null;
Action sendHeartbeat = () => { if (forward != null) forward(); };
//ugly, since it does not scale to multiple observers:
IObservable<Unit> heartbeatObs =
  Observable.FromEvent(handler => {
    forward = handler;
  }, _ => { forward = null; });
operation(sendHeartbeat);

即使我没有使用零维正向操作，而是使用它们的集合，这也会很难看，因为我将再次为EventHandlers实现+=和-=运算符。

Answer 1

我不确定我是否完全理解你的问题，但如果我理解了，你可以做这样的事情，它使用Subject来传输心跳。您可以在名称空间System.Reactive.Subjects中找到主题。

 // set up a subject
 Subject<Unit> heartbeatSubject=new Subject<Unit>();
 // set up the heartbeataction to push units down the subject
 Action sendHeartbeat = () => { heartbeatSubject.OnNext(Unit.Default); };
 // use the heartbeat subject somehow.
 heartbeatSubject.Subscribe(_=>Console.WriteLine("Operation produced a heartbeat"));
 // start the operation
 operation(sendHeartbeat);

编辑

正如你在评论中提到的，根据官方RX行，除了在演示/示例代码中，最好避免使用主题。我刚刚尝试过使用Observable.Create，哪个可能更好？

Action<Action> longRunningAction=(
    hb => { for(int i=0;i<100;i++) { hb(); Thread.Sleep(1000);}});
var ob=Observable.Create<Unit>(
(IObserver<Unit> observer) =>
{
    longRunningAction(()=>observer.OnNext(Unit.Default));
    return Disposable.Create(() => Console.WriteLine("Observer has unsubscribed"));
});
ob.Subscribe(_=>Console.WriteLine("Received a heartbeat"));