Subscribed unsubscribe Subscribe Subscribe

F# Monkey

about

F# on RX

f:id:callmekohei00:20170325142408p:plain

Prepare01 ( Library )

source https://www.nuget.org/api/v2

nuget FSharp.Control.Reactive
nuget system.reactive

Prepare02 ( required )

F#のコードにRxライブラリを取り込むのはこんな感じ

#r @"./packages/System.Reactive.Core/lib/net46/System.Reactive.Core.dll"
#r @"./packages/System.Reactive.Linq/lib/net46/System.Reactive.Linq.dll"
#r @"./packages/System.Reactive.Interfaces/lib/net45/System.Reactive.Interfaces.dll"
#r @"./packages/System.Reactive.PlatformServices/lib/net46/System.Reactive.PlatformServices.dll"
#r @"./packages/FSharp.Control.Reactive/lib/net45/FSharp.Control.Reactive.dll"

open System                  // System.EventArgs, System.Timers.ElapsedEventArgs
open System.Reactive.Linq    // IGroupedObservable
open FSharp.Control.Reactive // Observable methods

Rx overview

Rxの勘所〜

IObservable<'T> とは string とか List とかと同じタイプの一種

IObservable<string><string> とは IObservable の中身が string ということ

"Hello" という string を IObservable<string> にするには Observable.single "hello world"

Intro

Rxとはこんな感じ〜

// プリント関数
let oStdout ( obs:IObservable<'T> ) = 
    obs
    /// 全てのIObservableに対しての操作が終わったら`-----`を印字する
    |> Observable.finallyDo( fun _ -> stdout.WriteLine("-------------") )
    /// IObservableを印字する
    |> Observable.subscribe( fun x -> stdout.WriteLine(x) )


Observable.single "hello world"
|> oStdout

Observable.single "hello world"
|> Observable.map ( fun s -> s.ToUpper() )
|> oStdout

Observable.single "hello world"
|> Observable.concat ( Observable.single "callmekohei" )
|> oStdout

結果

hello world
-------------
HELLO WORLD
-------------
hello world
callmekohei
-------------

Event with Rx

IEvent<_>IObservable<_>と同じ感覚でオーケー

// イベントをインスタンス化
let e = Event<_>()

// イベントを実体化する
e.Publish
// イベントハンドラを登録する
// イベントハンドラとはイベントを実行するときに行う操作のこと
|> Observable.add( fun (e:string) -> stdout.WriteLine(e) )

// イベントを実行する
e.Trigger("hello world")

結果

hello world


IEvent<ElapsedEventHandler,ElapsedEventArgs>IObservable<_>と同じ感覚でオーケー

/// file name is timer.fsx
open System.Timers

// イベントをインスタンス化
let timer = new Timer(1000.)

/// イベントを実体化
timer.Elapsed
/// イベントハンドラを登録
|> Observable.add (fun (e:ElapsedEventArgs) -> printfn "%A"( e.SignalTime ))

// イベントを実行(1000msごとにイベントを実行する)
timer.Start()

stdout.WriteLine("please enter key...")

/// dispose method call
stdin.ReadLine()
stdout.WriteLine("dispose method call.")

/// イベントを破棄する(メモリクリアする)<--- これ重要!
timer.Disposed

結果

$ fsharpi timer.fsx 
please enter key...
3/25/2017 1:41:42 PM
3/25/2017 1:41:43 PM
3/25/2017 1:41:44 PM
3/25/2017 1:41:45 PM
3/25/2017 1:41:46 PM

dispose method call.

Observable methods

いろいろなメソッド〜

//----------------------------------------------------------------------------
// single, toObservable and generate
//

"hello world"
|> Observable.single
|> oStdout
/// hello world

["hello world"]
|> Observable.toObservable
|> oStdout
/// hello world

Observable.generate 0 (fun i -> i < 3) ( fun i -> i + 1 ) ( fun i -> i * i )
|> oStdout
/// 0
/// 1
/// 4


//----------------------------------------------------------------------------
// apply, defer
//

["hello world"]
|> Observable.toObservable
|> Observable.apply ( Observable.toObservable ( [fun s -> s.ToUpper()] ) )
|> oStdout
/// HELLO WORLD

fun _ -> Observable.single "hello world"
|> Observable.defer
|> oStdout
/// hello world


//----------------------------------------------------------------------------
// map, collect, flatmap and filter
//

"hello world"
|> Observable.single
|> Observable.map ( fun s -> s.ToUpper() )
|> oStdout
/// HELLO WORLD

["hello world"]
|> Observable.toObservable
|> Observable.thenMap ( fun s -> s.ToUpper() )
|> fun x -> [x]
|> Observable.joinWhen
|> oStdout
/// HELLO WORLD

["hello";"world"]
|> Observable.collect (fun s -> Observable.single ( s.ToUpper() ) )
|> oStdout
/// HELLO
/// WORLD

["hello";"world"]
|> Observable.toObservable
|> Observable.flatmap (fun s -> Observable.single ( s.ToUpper() ) )
|> oStdout
/// HELLO
/// WORLD

["hello";"world"]
|> Observable.toObservable
|> Observable.filter (fun s -> s = "hello" )
|> oStdout
/// hello


//----------------------------------------------------------------------------
// concat 
//

Observable.single "hello"
|> Observable.concat ( Observable.single "world" )
|> oStdout
// hello
// world


//----------------------------------------------------------------------------
// iter, finallyDo 
//

["hello";"world"]
|> Observable.toObservable
|> Observable.iter ( fun _ -> stdout.WriteLine("---------") )
|> oStdout
/// ---------
/// hello
/// ---------
/// world


["hello";"world"]
|> Observable.toObservable
|> Observable.finallyDo ( fun _ -> stdout.WriteLine("---------") )
|> oStdout
/// hello
/// world
/// ---------


//----------------------------------------------------------------------------
// reduce and fold 
//

["hello";"world"]
|> Observable.toObservable
|> Observable.reduce ( fun a b -> a + " " + b )
|> oStdout
/// hello world

["hello";"world"]
|> Observable.toObservable
|> Observable.fold ( fun acc s -> acc + " " + s |> fun s -> s.TrimStart(' ') ) ""
|> oStdout
/// hello world


//----------------------------------------------------------------------------
// zip
//

(Observable.single "hello", Observable.single "world")
||> Observable.zip
|>  oStdout
/// (hello, world)



//----------------------------------------------------------------------------
// repeatCount
//

["I";"want";"to";"eat";"icecream"]
|> Observable.toObservable
|> Observable.repeatCount 2
|> Observable.subscribe( fun (obs:IObservable<string>) ->
    obs |> Observable.subscribe( fun obs' -> stdout.WriteLine(obs') ) |> ignore
    () )

/// I
/// want
/// I
/// to
/// want
/// eat
/// to
/// icecream
/// eat
/// icecream


//----------------------------------------------------------------------------
// groupBy
//

["hello";"world";"callmekohei"]
|> Observable.toObservable
|> Observable.groupBy ( fun s -> s.Length < 6 )
|> Observable.subscribe ( fun (g:IGroupedObservable<bool,string>) -> stdout.WriteLine( string(g.Key) , g |> oStdout))
/// True
/// hello
/// world
/// False
/// callmekohei


//----------------------------------------------------------------------------
// others
//

["hello";"world"]
|> Observable.toObservable
|> Observable.head
|> oStdout
/// hello

["hello";"world"]
|> Observable.toObservable
|> Observable.first
|> oStdout
/// hello

["hello";"world"]
|> Observable.toObservable
|> Observable.last
|> oStdout
/// world

["hello";"world"]
|> Observable.toObservable
|> Observable.elementAt 1
|> oStdout
/// world

["hello";"world"]
|> Observable.toObservable
|> Observable.contains "hello"
|> oStdout
/// True

["hello";"world"]
|> Observable.toObservable
|> Observable.exists ( fun s -> s = "hello" )
|> oStdout
/// True

["hello";"world"]
|> Observable.toObservable
|> Observable.count 
|> oStdout
/// 2

["hello";"world";"hello";"world"]
|> Observable.toObservable
|> Observable.distinct 
|> oStdout
/// hello
/// world

Observable.range 1 10 
|> Observable.skip 3
|> Observable.take 3
|> oStdout
// 4
// 5
// 6

Observable.range 1 10
|> Observable.skipWhile( fun i -> i < 5 )
|> Observable.takeWhile( fun i -> i < 10 )
|> oStdout
// 5
// 6
// 7
// 8
// 9

Observable.range 1 10
|> Observable.maxOf
|> oStdout
/// 10


Observable.range 1 10
|> Observable.all ( fun n -> n % 2 = 0 )
|> oStdout
// False

Observable.empty 
|> Observable.any 
|> oStdout
// False


//----------------------------------------------------------------------------
// error raise
//

Observable.throw ( exn "エラーだよ" )
|> oStdout
/// エラーだよ

Observable methods ( async )

いろいろなメソッド〜 その2( async )




  
Remove all ads