F#でリアクティブプログラミング その２

F#でReactive Extensionsみたいなことが出来るということでじゃっかんハッスルしてたのですが、まだまだ提供されている便利メソッドは少ないというのが実情みたいです。（id:neueccさん情報）確かにObservableモジュールを眺めてみてもメソッドが数えるほどしかありません。これは今後拡充していくことに期待か、Reactive Exteensionsとがっちり統合するとか、そういう未来を期待したいですね！

とまぁ、現状ある範囲でも軽く遊ぶぶんには十分なのですが、個人的に気になるところがありました。これって例外処理どうやるの？？？ということで、以下のように3で割り切れる数のときは例外を出すようにちょろっとプログラムを変えて実行してみました。

// int型を発行するIEventを作る
let e = Event<int>()
let event = e.Publish
// 偶数のみ出力するようにフィルタリング
event
    |> Observable.filter (fun i -> 
                            match i with 
                            // 3で割り切れるときは例外
                            | i when i % 3 = 0 -> failwith "NG"
                            // 2で割り切れるときはOK
                            | i when i % 2 = 0 -> true
                            // それ以外はダメ
                            | _ -> false)
    |> Observable.subscribe (printfn "%d")
    |> ignore

// 1〜10までTrigger(発行)してみる
[1..10]
    |> Seq.iter e.Trigger

これを実行すると、エラーで落ちるのかな？と思ったら落ちないで、以下の結果になりました。

2
4
8
10

例外を完全にシカトしてます。ということでF# PowerPackのソースコードをひっぱってきてObservableのコードを見てみました。因みにF# PowerPackのコードをダウンロードしてcompiler\2.0\Nov2010\src\fsharp\FSharp.Coreの中にあるcontrol.fs内にコードがありました。Observable.subscribeを見てみると以下のように定義されていました。単純にIObservable<'T>.Subscribeに委譲しているだけです。

[<CompiledName("Subscribe")>]
let subscribe (f: 'T -> unit) (w: IObservable<'T>) = w.Subscribe(f)

でも、普通のIObservableのSubscribeはIObserverしか受け取らないはずなのに、ラムダ式を受け取ってます。これは、同じファイルで以下のようにIObservableが拡張されてました。

type IObservable<'Args> with 

    [<CompiledName("AddToObservable")>] // give the extension member a 'nice', unmangled compiled name, unique within this module
    member x.Add(f: 'Args -> unit) = x.Subscribe f |> ignore

    [<CompiledName("SubscribeToObservable")>] // give the extension member a 'nice', unmangled compiled name, unique within this module
    member x.Subscribe(f) = 
        x.Subscribe { new IObserver<'Args> with 
                          member x.OnNext(args) = f args 
                          member x.OnError(e) = () 
                          member x.OnCompleted() = () } 

なんというか・・・例外握り潰してくれてます！！OnErrorとOnCompletedを完全にシカトしてます。これはいかん！！ということで、例外とか処理したかったら自分でなんとかしないといけないっぽいです。ということでOnNextとOnErrorの2つのラムダ式を受け取るsubscribe2という関数を定義します。

open System
module Observable =
    // 引数でOnNext, OnErrorに対応するラムダ式を受け取る
    // 関数を定義しておく
    let subscribe2
        (s : 'T -> unit) 
        (e : Exception -> unit) 
        (o : IObservable<'T>) =
        o.Subscribe({ new IObserver<'T> with
                        member x.OnNext(args) = s(args)
                        member x.OnError(ex) = e(ex)
                        member x.OnCompleted() = ()})

そして、最初のプログラムをsubscribeから上記で定義したsubscribe2を使うように書き換えます。

// int型を発行するIEventを作る
let e = Event<int>()
let event = e.Publish
// 偶数のみ出力するようにフィルタリング
event
    |> Observable.filter (fun i -> 
                            match i with 
                            // 3で割り切れるときは例外
                            | i when i % 3 = 0 -> failwith "NG"
                            // 2で割り切れるときはOK
                            | i when i % 2 = 0 -> true
                            // それ以外はダメ
                            | _ -> false)
    // subscribe2に変更
    |> Observable.subscribe2
        (printfn "%d") // OnNext
        (printfn "%A") // OnError
    |> ignore

// 1〜10までTrigger(発行)してみる
[1..10]
    |> Seq.iter e.Trigger

これを実行すると、ちゃんとOnErrorも呼ばれます！！

2
System.Exception: NG
   場所 FSI_0004.clo@21-8.Invoke(Int32 i) 場所 c:\users\xxxx\documents\visual studio 2010\Projects\Library1\Library1\Script.fsx:行 24
   場所 Microsoft.FSharp.Control.ObservableModule.Filter@2263-2.Invoke(T x)
   場所 Microsoft.FSharp.Control.ObservableModule.Choose@2255-4.Next(T value)
4
System.Exception: NG
   場所 FSI_0004.clo@21-8.Invoke(Int32 i) 場所 c:\users\xxxx\documents\visual studio 2010\Projects\Library1\Library1\Script.fsx:行 24
   場所 Microsoft.FSharp.Control.ObservableModule.Filter@2263-2.Invoke(T x)
   場所 Microsoft.FSharp.Control.ObservableModule.Choose@2255-4.Next(T value)
8
System.Exception: NG
   場所 FSI_0004.clo@21-8.Invoke(Int32 i) 場所 c:\users\xxxx\documents\visual studio 2010\Projects\Library1\Library1\Script.fsx:行 24
   場所 Microsoft.FSharp.Control.ObservableModule.Filter@2263-2.Invoke(T x)
   場所 Microsoft.FSharp.Control.ObservableModule.Choose@2255-4.Next(T value)
10

うん！ちょっとすっきり。ここら辺とか標準のライブラリに含まれる日を夢見て。