我有一个 Action<Action> operation
它执行一些长时间运行的工作并通过提供的 Action
发送心跳争论。
在我调用它之前,我想设置一个 IObservable<Unit>
每次操作发送心跳时都会生成一个 Next 元素。
我该怎么做?
创建 IObservable
是not that complicated , 但我猜概念上最简单的方法是 create it from an event ,但遗憾的是您不能在方法体中创建事件(除非您知道一种方法)。我可以使用 Observable.FromEvent(Action<Action> addHandler, Action<Action> removeHandler)
,但这需要丑陋的封闭式 Actions 才能将方法调用转发给(参见下面的示例)。
有没有更优雅的方式?
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);
即使我不使用零维前向 Action ,而是使用它们的集合,它也会很丑陋,因为我会实现 +=
和 -=
EventHandlers
的运算符再来一遍。
最佳答案
不确定我是否完全理解您的问题,但如果我理解了,您可以做这样的事情,它使用 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"));
关于c# - 行动到事件,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/17828662/