Observable vs FSharpx asyncSeq

Question

我有下面的tweet流类。它有TweetReceived事件，可以与我的系统的其他组件一起使用。

它看起来还可以，但我觉得它比应该的更复杂。

有没有任何工具可以在不必自己实现mbox/event机制的情况下为我提供此功能？

此外，您是否建议使用asyncSeq而不是IObservable？

谢谢！

type TweetStream ( cfg:oauth.Config) =
let token = TwitterToken.Token (cfg.accessToken,
cfg.accessTokenSecret,
cfg.appKey, 
cfg.appSecret)
let stream = new SimpleStream("https://stream.twitter.com/1.1/statuses/sample.json")
let event = new Event<_>()
let agent = MailboxProcessor.Start(fun (mbox) ->
let rec loop () =
async {
let! msg = mbox.Receive()
do event.Trigger(msg)
return! loop()
}
loop ()) 
member x.TweetReceived = event.Publish
member x.Start () =
Task.Factory.StartNew(fun () -> stream.StartStream(token, agent.Post))
|> ignore
member x.Stop = stream.StopStream

更新：感谢托马斯对第二个问题的快速(一如既往)回答。

我的第一个问题可能有点不清楚，所以我重构了代码，使类AgentEvent可见，并重新表述了第一个问题：有没有一种方法可以更容易地实现AgentEvent中的逻辑？这种逻辑已经在某个地方实现了吗？

我之所以这么问，是因为这感觉像是一种常见的使用模式。

type AgentEvent<'t>()=
let event = new Event<'t>()
let agent = MailboxProcessor.Start(fun (mbox) ->
let rec loop () =
async {
let! msg = mbox.Receive()
do event.Trigger(msg)
return! loop()
}
loop ()) 
member x.Event = event.Publish
member x.Post = agent.Post
type TweetStream ( cfg:oauth.Config) =
let token = TwitterToken.Token (cfg.accessToken,
cfg.accessTokenSecret,
cfg.appKey, 
cfg.appSecret)
let stream = new SimpleStream("https://stream.twitter.com/1.1/statuses/sample.json")
let agentEvent = AgentEvent()
member x.TweetReceived = agentEvent.Event
member x.Start () =
Task.Factory.StartNew(fun () -> stream.StartStream(token, agentEvent.Post))
|> ignore
member x.Stop = stream.StopStream

Answer 1

我认为IObservable是发布事件的正确抽象。至于处理它们，我会使用反应扩展或F#代理(MailboxProcessor)，这取决于你想做什么

请注意，F#自动将事件表示为IObservable值(实际上是IEvent，但它继承自observable)，因此您可以直接在TweetReceived上使用Reactive Extensions。

什么是正确的表述？

• asyncSeq的要点是，它可以让你控制数据生成的速度-它就像async一样，你必须启动它才能真正完成工作并获得值-所以如果你可以启动一些操作(例如下载接下来的几个字节)来获得下一个值，这很有用

• 当你无法控制数据源时，IObservable很有用——当它只是不断产生值，而你无法暂停它时——这似乎更适合推特。

至于处理，我认为当反应式扩展已经实现了您需要的操作时，它们是很好的。当您需要编写一些自定义逻辑(在Rx中不容易表达)时，使用Agent是编写自己的Rx类函数的好方法。