IObservable订阅的自我纠错自我处理 [英] Self-error-dependent self-disposal of IObservable subscriptions

问题描述

我有F＃代码，如下所示：

I have F# code that looks like this:

module O = Control.Observable
//...
use ss = serve' 4000
         |> O.subscribe
            (fun c -> use cs = RSS.items
                               |> O.subscribe (bytes >> c.SendAll) |> ignore)

---

其中

---

where

serve'    : int -> IObservable<Socket>
c         : Socket
RSS.items : IObservable<XElement>
bytes     : XElement -> byte []
c.SendAll : byte [] -> unit

• 最惯用的方式是保留 cs 直到 c.SendAll 失败？


• 是否有可能定义 Observable.subscribeUntilError（action） 如果操作失败，订阅将被处理;否则操作只要 IObservable 持续推送就运行


• What is the most idiomatic way to retain cs until c.SendAll fails?
• Is there or is it possible to define Observable.subscribeUntilError(action) where if action fails, subscription gets disposed; otherwise action is run as long as IObservable keeps pushing?

推荐答案

我已经提出：

let inline Δ<'a> = Unchecked.defaultof<'a>
let inline LOG x = printf "%A" x

module O = Observable
  let d = ref (Δ:IDisposable)
  let b x = try a x with e -> LOG e; let s = !d in if s <> Δ then s.Dispose()
  d := o |> O.subscribe b
  {
    new IDisposable with
      member x.Dispose() = let s = !d in if s <> Δ then d := Δ; s.Dispose()
  }





---





为了证明差异，请尝试 main ！

使用 subscribe ：

use s = new Subject<int>()
use x = s |> O.subscribe (fun _ -> raise <| Exception ())
use y = s |> O.subscribe (printf "%i")
s.OnNext 20

应用程序崩溃down：

application comes crashing down:

Unhandled Exception: System.Exception: Exception of type 'System.Exception' was thrown.
   at Microsoft.FSharp.Core.Operators.Raise[T](Exception exn)
   at [e-mail](Int32 _arg1) in C:\Eniox\Eniox.News.Google\Eniox.News.Google\Program.fs:line 60
   at [e-mail]-IObserver`1-OnNext(T value)
   at System.Reactive.Observer`1.OnNext(T value)
   at System.Reactive.Subjects.Subject`1.OnNext(T value)
   at Program.System.main(String[] args) in C:\Eniox\Eniox.News.Google\Eniox.News.Google\Program.fs:line 606





---





现在使用 subscribeUE ：

---

Now using subscribeUE:

use s = new Subject<int>()
use x = s |> O.subscribeUE (fun _ -> raise <| Exception ())
use y = s |> O.subscribe   (printf "%i")
s.OnNext 20

愉快地配置订阅 x ，应用程序继续运行而不打嗝，正常退出！输出与 LOG =忽略：

delightfully disposes subscription x, application continues to run without a hiccup and exits normally! Output with LOG = ignore:

20





---





我很想知道是否可比功能存在于RX 2.0中的某个地方，因为我发现这个组合器太有用了，可以省略。

---

I would love to know whether comparable functionality actually exists somewhere in RX 2.0 as I find this combinator to be too useful to leave out.

这篇关于IObservable订阅的自我纠错自我处理的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！