f# observable fork 和副作用

Question

我有一个 observable 链，初始 observable 来自网络，每次消息准备好读取时都会被触发。然后下一个处理程序读取消息并反序列化它。现在我有一个 observable 的分支，一个是消息处理程序，另一个是记录消息。

问题是因为我使用 observable 我实际上会尝试阅读消息两次。

我知道使用 Event 而不是 Observable 可以解决问题，但是我会遇到垃圾收集问题，这可能会导致套接字无法被收集。

我想到的一个解决方案是插入某种分隔符，结束一个可观察链并创建一个新的，这样的函数是否已经作为 fsharp 或其他库的一部分存在。

这个问题还有其他解决方案吗？

编辑：

无法正常工作的代码示例

let messagesStream = 
  socket.observable |>
  Observable.map (fun () -> socket.read ()) |>
  Observable.map (fun m -> deserialize m)

messagesStream |> Observable.add (fun m -> printf m)
messagesStream |> Observable.add (fun m -> handle m)

Answer 1

添加一些日志记录的最简单方法是使用Observable.iter，如下所示：

let messagesStream = 
  socket.observable
  |> Observable.map (fun () -> socket.read ()) 
  |> Observable.map (fun m -> deserialize m)
  |> Observable.iter (printfn "%A")

messagesStream |> Observable.add (fun m -> handle m)  

Answer 2

听起来您可以创建一个 observable 来处理从网络读取消息并将其反序列化。假设这是使用标准 Rx 运算符完成的，则应该返回一个推送新的反序列化网络消息的 observable。

您可以有 2 个订阅者订阅该 observable，一个订阅者使用您想要的任何业务逻辑对新消息做出反应，第二个订阅者记录消息。

这应该消除从网络多次读取的副作用。推送 2 个反序列化消息的副本不会产生副作用。