{"id":882,"date":"2009-10-22T17:23:52","date_gmt":"2009-10-22T15:23:52","guid":{"rendered":"http:\/\/www.navision-blog.de\/2009\/10\/22\/iobservableiobserver-using-the-reactive-framework-with-f-part-ii\/"},"modified":"2009-10-22T17:32:06","modified_gmt":"2009-10-22T15:32:06","slug":"iobservable-iobserver-using-the-reactive-framework-with-fsharp-part-ii","status":"publish","type":"post","link":"http:\/\/www.navision-blog.de\/blog\/2009\/10\/22\/iobservable-iobserver-using-the-reactive-framework-with-fsharp-part-ii\/","title":{"rendered":"IObservable\/IObserver – Using the Reactive Framework with F# – part II"},"content":{"rendered":"

In the last article<\/a> I showed how to filter and combine events via the Reactive Framework and how to deal with errors. This time we will create our own observables. <\/p>\n

I got the idea for this sample from a very good Expert to Expert video (\u201cReactive Framework (Rx) Under the Hood<\/a>\u201d) with Erik Meijer and Wes Dyer.<\/p>\n

We want to implement an asynchronous dictionary lookup. Whenever sometimes types something into the Textbox, our application starts looking into a dictionary and searches for words starting with the given prefix. <\/p>\n

\"Async<\/p>\n

Let’s start with generating this simple form:<\/p>\n

\n

open<\/span> System.Windows.Forms<\/p>\n

 <\/p>\n

let<\/span> form = new<\/span> Form(Visible=true<\/span>, Text="Async dict"<\/span>, <\/p>\n

                       TopMost=true<\/span>) <\/p>\n

let<\/span> textBox1 = <\/p>\n

  new<\/span> TextBox(<\/p>\n

    Location = new<\/span> System.Drawing.Point(12, 12),<\/p>\n

    Size = new<\/span> System.Drawing.Size(260, 20))<\/p>\n

form.Controls.Add textBox1<\/p>\n

 <\/p>\n

let<\/span> resultsBox = <\/p>\n

  new<\/span> ListBox(<\/p>\n

    Location = new<\/span> System.Drawing.Point(13, 39),<\/p>\n

    Size = new<\/span> System.Drawing.Size(259, 211))<\/p>\n

form.Controls.Add resultsBox<\/p>\n

 <\/p>\n

\/\/ create a list with common words<\/span><\/p>\n

\/\/ this might be very large<\/span><\/p>\n

let<\/span> data =<\/p>\n

  ["hell"<\/span>; "Hello"<\/span>; "Halle"<\/span>; "Html"<\/span>; <\/p>\n

      "Bonn"<\/span>; "Bonjour"<\/span>; "Steffen"<\/span>]  <\/p>\n

 <\/p>\n

\/\/ create observable for text changes<\/span><\/p>\n

let<\/span> textChanged =<\/p>\n

  textBox1.TextChanged<\/p>\n

    |> Observable.map (fun<\/span> _ -><\/span> textBox1.Text)<\/p>\n<\/p><\/div>\n

Now we have to define a base class for observables. This class will help our dictionary lookup function to use the IObservable<T> interface:<\/p>\n

\n

module<\/span> Observable<\/p>\n

 <\/p>\n

\/\/\/ A Observable base class which notifies <\/span><\/p>\n

\/\/\/ all observers in parallel  <\/span><\/p>\n

type<\/span> ‘a Observable() =<\/p>\n

  let<\/span> mutable<\/span> observers = []<\/p>\n

 <\/p>\n

  \/\/\/ Notifies all observers in parallel about the new value<\/span><\/p>\n

  let<\/span> notifyObservers f =<\/p>\n

    observers<\/p>\n

      |> Seq.map (fun<\/span> (observer:IObserver<‘a>) \u2013><\/span> <\/p>\n

                         async { return<\/span> f observer})<\/p>\n

      |> Async.Parallel<\/p>\n

      |> Async.RunSynchronously<\/p>\n

      |> ignore<\/p>\n

 <\/p>\n

  interface<\/span> IObservable<‘a> with<\/span><\/p>\n

    member<\/span> observable.Subscribe(observer)  =<\/p>\n

      \/\/ subscribe observer<\/span><\/p>\n

      observers <- observer :: observers<\/p>\n

      \/\/ create Disposable to unsubscribe observer later<\/span><\/p>\n

      {new<\/span> IDisposable with<\/span><\/p>\n

         member<\/span> this.Dispose() = <\/p>\n

            observers <- <\/p>\n

                observers |> List.filter ((<>) observer)}<\/p>\n

 <\/p>\n

  \/\/\/ Notifies all observers in parallel about the new value<\/span><\/p>\n

  member<\/span> observable.OnNext value = <\/p>\n

   notifyObservers (fun<\/span> observer -><\/span> observer.OnNext value)<\/p>\n

 <\/p>\n

  \/\/\/ Notifies all observers in parallel about the error<\/span><\/p>\n

  \/\/\/ and finishes all observations<\/span><\/p>\n

  member<\/span> observable.OnError error = <\/p>\n

    notifyObservers (fun<\/span> observer -><\/span> observer.OnError error)<\/p>\n

    observers <- []<\/p>\n

 <\/p>\n

  \/\/\/ Notifies all observers in parallel about the completion<\/span><\/p>\n

  \/\/\/ and finishes all observations<\/span><\/p>\n

  member<\/span> observable.Completed = <\/p>\n

    notifyObservers (fun<\/span> observer -><\/span> observer.OnCompleted())<\/p>\n

    observers <- []  <\/p>\n<\/p><\/div>\n

I hope there will be a similar base class in the .NET Framework 4.0 RTM. <\/p>\n

Now we are able to use this class and to build our dictionary lookup observable:<\/p>\n

\n

let<\/span> wordsObservable = new<\/span> Observable.Observable<_>()<\/p>\n

 <\/p>\n

let<\/span> findWords prefix =<\/p>\n

  if<\/span> prefix <> ""<\/span> then<\/span><\/p>\n

    let<\/span> prefix’ = prefix.ToUpper()<\/p>\n

    for<\/span> word in<\/span> data do<\/span><\/p>\n

      if<\/span> word.ToUpper().StartsWith(prefix’) then<\/span> <\/p>\n

        wordsObservable.OnNext (prefix,word)<\/p>\n<\/p><\/div>\n

The last step is to create observers and subscribe them to the observables:<\/p>\n

\n

\/\/ create observers<\/span><\/p>\n

let<\/span> clean = <\/p>\n

  textChanged <\/p>\n

    |> Observable.subscribe (fun<\/span> _ -><\/span> resultsBox.Items.Clear())<\/p>\n

 <\/p>\n

let<\/span> searchForWords =<\/p>\n

  \/\/ Every time the text changes <\/span><\/p>\n

  \/\/ we start our wordsObservable to push words<\/span><\/p>\n

  textChanged <\/p>\n

    |> Observable.subscribe (fun<\/span> text -><\/span> findWords text)<\/p>\n

 <\/p>\n

let<\/span> wordFound =<\/p>\n

  \/\/ subscribe to the "word found"-event<\/span><\/p>\n

  wordsObservable    <\/p>\n

    |> Observable.subscribe <\/p>\n

        (fun<\/span> (_,word) -><\/span> resultsBox.Items.Add word |> ignore)<\/p>\n<\/p><\/div>\n","protected":false},"excerpt":{"rendered":"

In the last article I showed how to filter and combine events via the Reactive Framework and how to deal with errors. This time we will create our own observables. I got the idea for this sample from a very good Expert to Expert video (\u201cReactive Framework (Rx) Under the Hood\u201d) with Erik Meijer and […]<\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":[],"categories":[448],"tags":[664,580,579,578],"_links":{"self":[{"href":"http:\/\/www.navision-blog.de\/blog\/wp-json\/wp\/v2\/posts\/882"}],"collection":[{"href":"http:\/\/www.navision-blog.de\/blog\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"http:\/\/www.navision-blog.de\/blog\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"http:\/\/www.navision-blog.de\/blog\/wp-json\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"http:\/\/www.navision-blog.de\/blog\/wp-json\/wp\/v2\/comments?post=882"}],"version-history":[{"count":6,"href":"http:\/\/www.navision-blog.de\/blog\/wp-json\/wp\/v2\/posts\/882\/revisions"}],"predecessor-version":[{"id":888,"href":"http:\/\/www.navision-blog.de\/blog\/wp-json\/wp\/v2\/posts\/882\/revisions\/888"}],"wp:attachment":[{"href":"http:\/\/www.navision-blog.de\/blog\/wp-json\/wp\/v2\/media?parent=882"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"http:\/\/www.navision-blog.de\/blog\/wp-json\/wp\/v2\/categories?post=882"},{"taxonomy":"post_tag","embeddable":true,"href":"http:\/\/www.navision-blog.de\/blog\/wp-json\/wp\/v2\/tags?post=882"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}