f# - 如何创建 "subject"F# Observable？

Question

在 Rx 中，我可以像这样创建一个主题(某物):

let s = new Subject<int>()

let subscription = 
  s.Subscribe (fun x -> 
    Console.WriteLine x
  )

s.OnNext 123

主体既是可观察的又是观察者。这在将回调转换为流时很有用。

使用 F# 的内置 Observable 模块有什么等价物？

Answer 1

F# Snippets 上有一个示例实现:

Observable.Subject

The Subject type implements both IObserver and IObservable. It is functionally equivalent to the type of the same name in the Reactive Extensions (Rx) library.

module Observable

open System
open System.Collections.Generic

type Subject<'T> () =
   let sync = obj()
   let mutable stopped = false
   let observers = List<IObserver<'T>>()
   let iter f = observers |> Seq.iter f
   let onCompleted () =
      if not stopped then
         stopped <- true
         iter (fun observer -> observer.OnCompleted())
   let onError ex () =
      if not stopped then
         stopped <- true
         iter (fun observer -> observer.OnError(ex))
   let next value () =
      if not stopped then
         iter (fun observer -> observer.OnNext(value))
   let remove observer () =
      observers.Remove observer |> ignore
   member x.Next value = lock sync <| next value
   member x.Error ex = lock sync <| onError ex
   member x.Completed () = lock sync <| onCompleted
   interface IObserver<'T> with
      member x.OnCompleted() = x.Completed()
      member x.OnError ex = x.Error ex
      member x.OnNext value = x.Next value
   interface IObservable<'T> with
      member this.Subscribe(observer:IObserver<'T>) =
         observers.Add observer
         { new IDisposable with
            member this.Dispose() =
               lock sync <| remove observer
         }

do  let s = Subject()
    use d = s.Subscribe(fun x -> sprintf "%d" x |> Console.WriteLine)
    [1..12] |> Seq.iter s.Next