我有一个类负责在频繁但不规则的时间间隔内生成事件,其他类必须消耗并操作这些事件。我想使用Reactive Extensions来完成这个任务。
消费者方面非常简单;我的消费者类实现了
直接实现
消费者方面非常简单;我的消费者类实现了
IObserver<Payload>
,一切看起来都很好。问题出现在生产者类上。直接实现
IObservable<Payload>
(也就是说,为IDisposable Subscribe(IObserver<Payload> )
提供自己的实现)根据文档不推荐。它建议使用Observable.Create()
函数系列进行组合。由于我的类将运行很长时间,我尝试使用var myObservable = Observable.Never()
创建一个Observable,然后在有新的Payloads可用时调用myObservable.Publish(payloadData)
。然而,当我这样做时,似乎没有触发我的消费者中的OnNext
实现。
我认为,作为一种解决方法,我可以在我的类中创建一个事件,然后使用FromEvent
函数创建Observable,但这似乎是一种过于复杂的方法(即,新的Observables“需要”事件工作似乎很奇怪)。这里有没有我忽视的简单方法?创建自己的Observable源的“标准”方式是什么?
IObservable<T>
或者IObserver<T>
中的任意一个,那么很可能你做错了什么。 - Enigmativity