增廣Observable.Create爲F＃

Question

我想增加

public static IObservable<TSource> Create<TSource>(
    Func<IObserver<TSource>, Action> subscribe) 
{...} 

爲了在F＃中使用，使得而不是一個函數或動作叫我就可以使用標準的F＃類型即IObserver -> (unit -> unit)。

我該如何做到這一點？

編輯：

添加完整的示例。不知道爲什麼obsAction不起作用。

open System 
open System.Reactive 
open System.Reactive.Disposables 
open System.Reactive.Linq 


type Observable with 
    static member Create(subscribe) = 
    Observable.Create(fun observer -> Action(subscribe observer)) 

let obsDispose (observer:IObserver<_>) = 
    let timer = new System.Timers.Timer() 
    timer.Interval <- 1000.00 
    let handlerTick = new Timers.ElapsedEventHandler(fun sender args ->  observer.OnNext("tick")) 
    let handlerElapse = new Timers.ElapsedEventHandler(fun sender args -> printfn "%A" args.SignalTime) 
    timer.Elapsed.AddHandler(handlerTick) 
    timer.Elapsed.AddHandler(handlerElapse) 
    timer.Start() 
    Disposable.Empty 

let obsAction (observer:IObserver<_>) = 
    let timer = new System.Timers.Timer() 
    timer.Interval <- 1000.00 
    let handlerTick = new Timers.ElapsedEventHandler(fun sender args -> observer.OnNext("tick")) 
    let handlerElapse = new Timers.ElapsedEventHandler(fun sender args -> printfn "%A" args.SignalTime) 
    timer.Elapsed.AddHandler(handlerTick) 
    timer.Elapsed.AddHandler(handlerElapse) 
    timer.Start() 
    let action() = 
     timer.Elapsed.RemoveHandler(handlerTick) 
     timer.Elapsed.RemoveHandler(handlerElapse) 
     timer.Dispose() 
    action 

let obsOtherAction (observer:IObserver<_>) = 
    let timer = new System.Timers.Timer() 
    timer.Interval <- 1000.00 
    let handlerTick = new Timers.ElapsedEventHandler(fun sender args -> observer.OnNext("tick")) 
    let handlerElapse = new Timers.ElapsedEventHandler(fun sender args -> printfn "%A" args.SignalTime) 
    timer.Elapsed.AddHandler(handlerTick) 
    timer.Elapsed.AddHandler(handlerElapse) 
    timer.Start() 
    new System.Action(fun() -> 
     timer.Elapsed.RemoveHandler(handlerTick) 
     timer.Elapsed.RemoveHandler(handlerElapse) 
     timer.Dispose()) 

let worksNeverStops = obsDispose |> Observable.Create |> Observable.subscribe(fun time -> printfn "Time: %A" time) 
let actionWorks = obsOtherAction |> Observable.Create |> Observable.subscribe(fun time -> printfn "Time: %A" time) 
let doesNotWork = obsAction |> Observable.Create |> Observable.subscribe(fun time -> printfn "Time: %A" time) 

Answer 1

您面臨的問題是FP陷阱。

在，

static member Create(subscribe) = 
    Observable.Create(fun observer -> Action(subscribe observer)) 

的subscribe類型是IObserver<_> -> unit -> unit。

現在有一個IObserver<_> -> unit -> unit 和IObserver<_> -> Action其中Action : unit -> unit之間的微妙差異。不同的是前者是咖喱，而後者則不是。

當觀察者訂閱，subscribe observer返回到()可以適用於獲得unit的方法 - 你的訂閱方法永遠不會真正被調用，直到應用的最後() - 這不會是直到取消訂閱由這點它已經被分離了。

你可以迫使它克服它不被令行禁止：

let action() = ... | let action = (subscribe observer) 
Action(action) 

另外：

如果檢查IL，等效VB（功能裁判在VB中有更清晰的）版本對於

static member Create(subscribe) = 
    Observable.Create(fun observer -> Action(subscribe observer)) 

產生的Invoke爲FastFunc是：

Friend Function Invoke(ByVal arg As IObserver(Of a)) As Action 
    Return New Action(AddressOf New Observable-Create-Static@27-1(Of a)(Me.subscribe, arg).Invoke) 
End Function 

和：

static member Create(subscribe) = 
    Observable.Create(fun observer -> 
     let action = subscribe observer 
     Action(action)) 

是：

Friend Function Invoke(ByVal arg As IObserver(Of a)) As Action 
    Return New Action(AddressOf New Observable-Create-Static@28-1(Me.subscribe.Invoke(arg)).Invoke) 
End Function 

1. AddressOf New Closure(Me.subscribe, arg).Invoke - >訂閱的功能將不會被調用，直至處置動作被調用。

2. AddressOf New Closure(Me.subscribe.Invoke(arg)).Invoke - >訂閱的功能實際上被調用和按預期的方式返回所產生的作用。

我希望現在清楚爲什麼第二種情況起作用，而不是第一種。