Rash thoughts about .NET, C#, F# and Dynamics NAV.


"Every solution will only lead to new problems."

Monday, 21. December 2009


Observing asynchronous downloads with F# and the Reactive Extensions for .NET

Filed under: F# — Steffen Forkmann at 18:01 Uhr

In one of my lasts posts I showed how we can transform some of the operators from the Reactive Framework for an easier use in F#. This time I will demonstrate how we can use this API and asynchronous workflows to download a couple of websites asynchronous and in parallel.

First of all we create a synchronous download function:

let SyncHttp (url:string) =

  let request = WebRequest.Create url

  let response = request.GetResponse()

  use stream = response.GetResponseStream()

  use reader = new StreamReader(stream)

  let result = reader.ReadToEnd()

  url,result

In F# we can easily make this function asynchronously:

let AsyncHttp (url:string) =

  async { // using asynchronous workflows

    let request = WebRequest.Create url

    // async call of GetResponse

    let! response = request.AsyncGetResponse()

    use stream = response.GetResponseStream()

    use reader = new StreamReader(stream)

    // async call of ReadToEnd

    let! result = reader.AsyncReadToEnd()

    return url,result} 

Now we use this AsyncHttp function to create a list of async downloads:

/// Async<string * string> list

let downloads =

  ["http://www.google.com";

   "http://www.twitter.com";

   "http://www.nytimes.com/";

   "http://www.navision-blog.de/";

   "http://www.nba.com/"

    |> List.map AsyncHttp

The next step is to convert the list of async calls into a list of observables and to merge this list into a single IObservable. The effect is that whenever one download is completed we will be notified about the result:

/// IObservable<string * string>

let observableDownloads =

  downloads

    |> Seq.map Observable.ofAsync

    |> Observable.merge

 

 

observableDownloads

  |> Observable.subscribe (fun (url,result) ->

       printfn "%A: %d" url result.Length)

 

If you want to learn more about Observable.merge and see the marble diagram you should watch this video.

Tags: , ,

Friday, 27. November 2009


Observing the FileSystem

Filed under: Diverses — Steffen Forkmann at 10:38 Uhr

The following class is a wrapper for the System.IO.FileSystemWatcher and converts the FileSystem events into observables. You need to download and reference the Reactive Extensions for .NET (Rx) to use this code:

public class FileSystemObservable

{

  private readonly FileSystemWatcher _fileSystemWatcher;

 

 

  public FileSystemObservable(string directory,

    string filter, bool includeSubdirectories)

  {

    _fileSystemWatcher =

      new FileSystemWatcher(directory, filter)

        {

          EnableRaisingEvents = true,

          IncludeSubdirectories = includeSubdirectories

        };

 

 

    ChangedFiles =

      Observable.FromEvent<FileSystemEventHandler,

                             FileSystemEventArgs>

      (h => h.Invoke,

       h => _fileSystemWatcher.Changed += h,

       h => _fileSystemWatcher.Changed -= h)

      .Select(x => x.EventArgs);

 

    CreatedFiles =

      Observable.FromEvent<FileSystemEventHandler,

                             FileSystemEventArgs>

      (h => h.Invoke,

       h => _fileSystemWatcher.Created += h,

       h => _fileSystemWatcher.Created -= h)

      .Select(x => x.EventArgs);

 

    DeletedFiles =

      Observable.FromEvent<FileSystemEventHandler,

                             FileSystemEventArgs>

      (h => h.Invoke,

       h => _fileSystemWatcher.Deleted += h,

       h => _fileSystemWatcher.Deleted -= h)

      .Select(x => x.EventArgs);

 

    RenamedFiles =

      Observable.FromEvent<RenamedEventHandler,

                             RenamedEventArgs>

      (h => h.Invoke,

       h => _fileSystemWatcher.Renamed += h,

       h => _fileSystemWatcher.Renamed -= h)

      .Select(x => x.EventArgs);

 

    Errors =

      Observable.FromEvent<ErrorEventHandler, ErrorEventArgs>

      (h => h.Invoke,

       h => _fileSystemWatcher.Error += h,

       h => _fileSystemWatcher.Error -= h)

      .Select(x => x.EventArgs);

  }

 

  /// <summary>

  /// Gets or sets the errors.

  /// </summary>

  /// <value>The errors.</value>

  public IObservable<ErrorEventArgs> Errors

           { get; private set; }

 

  /// <summary>

  /// Gets the changed files.

  /// </summary>

  /// <value>The changed files.</value>

  public IObservable<FileSystemEventArgs> ChangedFiles

           { get; private set; }

 

  /// <summary>

  /// Gets the created files.

  /// </summary>

  /// <value>The created files.</value>

  public IObservable<FileSystemEventArgs> CreatedFiles

           { get; private set; }

 

  /// <summary>

  /// Gets the deleted files.

  /// </summary>

  /// <value>The deleted files.</value>

  public IObservable<FileSystemEventArgs> DeletedFiles

           { get; private set; }

 

  /// <summary>

  /// Gets the renamed files.

  /// </summary>

  /// <value>The renamed files.</value>

  public IObservable<RenamedEventArgs> RenamedFiles

           { get; private set; }

}

Now we can use the observable and can easily create meta-events:

IDisposable writer =

    new FileSystemObservable(@"d:\Test\", "*.*", false)

        .CreatedFiles

        .Where(x => (new FileInfo(x.FullPath)).Length > 0)

          // … you can do much more with the combinators

        .Select(x => x.Name)

        .Subscribe(Console.WriteLine);

Tags: ,

Tuesday, 24. November 2009


Generating an IObservable<T> from an IEvent in F#

Filed under: F# — Steffen Forkmann at 12:39 Uhr

Yesterday I showed how we can map some of the Rx operators to an API which looks more like the F# base classes. Today I wanted to use these mapped operators in a WPF-application written in F#.

F# gives us a nice way to use events as first class citizen (via IEvent) but these events implement their own version of IObservable<T> (in FSharp.Core.dll), which is unfortunately incompatible with the Rx version and therefore with the mapped API.

The solution I found is to wrap the F# IEvent with a Rx IObservable<T>:

/// Generates an observable from an IEvent

let fromEvent (event:IEvent<_,_>) =      

  Observable.Create<_>(fun x ->

    event.Subscribe x.OnNext |> ignore

    new System.Action(fun () -> ()))     

Now we are able to use the WPF events as observables:

// Register ListBox Commands

listBox1.KeyDown

  |> Observable.fromEvent

  |> Observable.filter (fun args -> args.Key = Key.Delete)

  |> Observable.subscribe deleteElement

I am interested if someone has a different and maybe better solution to this problem.

Updated: 21.12.2009 – Observable.Context is no longer supported by Rx ==> Removed

Tags: ,

Monday, 23. November 2009


Mapping the Reactive Framework (Rx) operators for F#

Filed under: F# — Steffen Forkmann at 13:35 Uhr

The “Reactive Extensions for .NET (Rx)” comes with lot’s of operators for using IObservable<T>. This code mimics the signature of the default F# sequence combinators and allows to use observables like sequences. It is a similar approach like Matthews Podwysocki’s blog post about mapping the IParallelEnumerable.

I will update this post from time to time to include more of the operators.

  • Update: 25.11.2009 – new operators mapped
  • Update: 21.11.2009 – Updated to new Rx release

module RxExtensions.Observable

 

open System.Linq

open System

open System.Threading

open System.Windows.Threading

 

type ‘a observable = IObservable<’a>

type ‘a observer = IObserver<’a>

 

/// converts a lambda in a System.Action

let asAction f = new System.Action(f)

 

/// System.Action whichs does nothing

let doNothing = asAction (fun () -> ())

 

/// Creates an observer

let createObserver next error completed =

  {new System.IObserver<_> with

      member this.OnCompleted() = completed()

      member this.OnError(e) = error e

      member this.OnNext(args) = next args}

 

/// Creates a new observable

let create f =

  Observable.Create<_>(fun x ->

    f x

    doNothing)  

 

/// Creates a observable from a async

let ofAsync async =

  create

    (fun obs ->

       Async.StartWithContinuations

         (async,obs.OnNext,obs.OnError,obs.OnError))

 

/// Gets a dispatcher Schdeuler for the current dispatcher

let getDispatcherScheduler _ =

  new DispatcherScheduler(Dispatcher.CurrentDispatcher)

 

/// Generates an observable from an IEvent

let fromEvent (event:IEvent<_,_>) = create (fun x -> event.Add x.OnNext)

 

/// Generates an empty observable

let empty<’a> = Observable.Empty<’a>()

 

/// Takes the head of the elements

let head = Observable.First

 

/// Merges the two observables

let mergeWith obs1 obs2 = Observable.Merge(obs2, obs1)

 

/// Merges all observables

let mergeAll (observables:IObservable<IObservable<’a>>) =

  Observable.Merge observables

 

/// Merges all observables

let merge (observables:(IObservable<’a>) seq) =

  Observable.Merge observables

 

/// Creates a range as an observable

let range start count = Observable.Range(start, count)

 

/// Converts a seq in an observable

let toObservable (seq: ‘a seq) = Observable.ToObservable seq

 

/// Converts a observable in a seq

let toEnumerable = Observable.ToEnumerable

 

/// Subscribes to the Observable with all 3 callbacks

let subscribeComplete next error completed (observable: ‘a observable) =

   observable.Subscribe(

     (fun x -> next x),

     (fun e -> error e),

     (fun () -> completed()))

 

/// Subscribes to the Observable with a

/// next and an error-function

let subscribeWithError next error observable =

  subscribeComplete next error (fun () -> ()) observable

 

/// Subscribes to the Observable with just a next-function

let subscribe next observable =

  subscribeWithError next ignore observable

 

/// throttles the observable for the given interval

let throttle interval observable =

  Observable.Throttle(observable,interval)   

 

/// throttles the observable scheduled on the current dispatcher

let throttleOnCurrentDispatcher interval observable =

  Observable.Throttle(

     observable,getDispatcherScheduler(),interval)

 

/// samples the observable at the given interval

let sample interval observable =

  Observable.Sample(observable,interval)   

 

/// samples the observable at the given interval

/// scheduled on the current dispatcher

let sampleOnCurrentDispatcher interval observable =

  Observable.Sample(

    observable,getDispatcherScheduler(),interval)

 

/// returns the observable sequence that reacts first.

let takeFirstOf2Reactions obs1 obs2 =

  Observable.Amb(obs1,obs2)

 

/// returns the observable sequence that reacts first.

let amb (obs: IObservable<’a> seq) =

  Observable.Amb obs  

 

/// returns the observable sequence that reacts first.

let takeFirstReaction (obs: IObservable<’a> seq) =

  Observable.Amb obs  

 

/// Matches when both observable sequences

/// have an available value.

let both obs1 obs2 = Observable.And(obs1,obs2)

 

/// Merges two observable sequences

/// into one observable sequence.

let zip obs1 obs2 =   

   Observable.Zip(obs1, obs2, Func<_,_,_>(fun a b -> a, b))

 

/// Merges two observable sequences into one observable sequence

/// whenever one of the observable sequences has a new value.

///    ==> More results than zip

let combineLatest obs1 obs2 =   

   Observable.CombineLatest(

     obs1, obs2, Func<_,_,_>(fun a b -> a, b))    

 

/// Concats the two observables to one observable

let concat observable =

  Observable.SelectMany(

    observable,

    Func<_,_>(fun (x:IObservable<’a>) -> x))

 

/// maps the given observable with the given function

let map f observable =

  Observable.Select(observable,Func<_,_>(f))  

 

/// maps the given observable with the given function

let mapi f observable =

  Observable.Select(observable,Func<_,_,_>(fun x i ->f i x))  

 

/// Filters all elements where the given predicate is satified

let filter f observable =

  Observable.Where(observable, Func<_,_>(f))

 

/// Splits the observable into two observables

/// Containing the elements for which the predicate returns

/// true and false respectively

let partition predicate observable =

  filter predicate observable,

  filter (predicate >> not) observable

 

/// Skips n elements

let skip n observable = Observable.Skip(observable, n)

 

/// Skips elements while the predicate is satisfied

let skipWhile f observable =

  Observable.SkipWhile(observable, Func<_,_>(f))

 

/// Runs all observable sequences in parallel

/// and combines their first values.

let forkJoin (observables: (‘a observable) seq) =

  Observable.ForkJoin observables

 

/// Counts the elements

let length = Observable.Count

 

/// Takes n elements

let take n observable =

  Observable.Take(observable, n)  

 

/// Determines whether the given observable is empty 

let isEmpty observable = Observable.IsEmpty observable

 

/// Determines whether the given observable is not empty 

let isNotEmpty observable = not (Observable.IsEmpty observable)

 

/// Determines whether an observable sequence

/// contains a specified value

/// which satisfies the given predicate

let exists predicate observable =

  observable

    |> skipWhile (predicate >> not)

    |> isNotEmpty

 

/// Continues an observable sequence that is terminated

/// by an exception with the next observable sequence.

let catch (newObservable:IObservable<’a>) failingObservable =

  Observable.Catch(failingObservable,newObservable) 

 

/// Takes elements while the predicate is satisfied

let takeWhile f observable =

  Observable.TakeWhile(observable, Func<_,_>(f))

 

/// Iterates through the observable

/// and performs the given side-effect

let perform f observable =

  Observable.Do(observable,fun x -> f x)

 

/// Invokes finallyAction after source observable

/// sequence terminates normally or by an exception.

let performFinally f observable =

  Observable.Finally(observable,fun _ -> f())

 

/// Folds the observable

let fold f seed observable =

  Observable.Aggregate(observable, seed, Func<_,_,_>(f))  

 

/// Retruns an observable from a async pattern 

let fromAsync beginF endF =

   Observable.FromAsyncPattern<_>(

     Func<_,_,_>(fun x y -> beginF(x,y)),

       (fun x -> endF x)).Invoke()

 

/// Runs all observable sequences in parallel

/// and combines their first values.

let subscribeAll next observables =

  observables |> Seq.map (subscribe next) |> Seq.toList     

 

type IObservable<’a> with

  /// Subscribes to the Observable with just a next-function

  member this.Subscribe(next) =

    subscribe next this

 

  /// Subscribes to the Observable with a next

  /// and an error-function

  member this.Subscribe(next,error) =

    subscribeWithError next error this

 

  /// Subscribes to the Observable with all 3 callbacks

  member this.Subscribe(next,error,completed) =

    subscribeComplete next error completed this

 

open System.Net

 

type WebRequest with

  member this.GetRequestStreamAsync() =

    fromAsync

     this.BeginGetRequestStream

     this.EndGetRequestStream

 

  member this.GetResponseAsync() =

    fromAsync

      this.BeginGetResponse

      this.EndGetResponse

 

  member this.GetResponseStreamAsync() =

    fromAsync

      this.BeginGetRequestStream

      this.EndGetRequestStream

 

type Async<’a> with

  member this.ToObservable() = ofAsync this

Tags: , , ,

Thursday, 22. October 2009


IObservable/IObserver – Using the Reactive Framework with F# – part II

Filed under: F# — Steffen Forkmann at 17:23 Uhr

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 (“Reactive Framework (Rx) Under the Hood”) with Erik Meijer and Wes Dyer.

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.

Async Dict

Let’s start with generating this simple form:

open System.Windows.Forms

 

let form = new Form(Visible=true, Text="Async dict",

                       TopMost=true)

let textBox1 =

  new TextBox(

    Location = new System.Drawing.Point(12, 12),

    Size = new System.Drawing.Size(260, 20))

form.Controls.Add textBox1

 

let resultsBox =

  new ListBox(

    Location = new System.Drawing.Point(13, 39),

    Size = new System.Drawing.Size(259, 211))

form.Controls.Add resultsBox

 

// create a list with common words

// this might be very large

let data =

  ["hell"; "Hello"; "Halle"; "Html";

      "Bonn"; "Bonjour"; "Steffen"

 

// create observable for text changes

let textChanged =

  textBox1.TextChanged

    |> Observable.map (fun _ -> textBox1.Text)

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

module Observable

 

/// A Observable base class which notifies

/// all observers in parallel 

type ‘a Observable() =

  let mutable observers = []

 

  /// Notifies all observers in parallel about the new value

  let notifyObservers f =

    observers

      |> Seq.map (fun (observer:IObserver<’a>) –>

                         async { return f observer})

      |> Async.Parallel

      |> Async.RunSynchronously

      |> ignore

 

  interface IObservable<’a> with

    member observable.Subscribe(observer)  =

      // subscribe observer

      observers <- observer :: observers

      // create Disposable to unsubscribe observer later

      {new IDisposable with

         member this.Dispose() =

            observers <-

                observers |> List.filter ((<>) observer)}

 

  /// Notifies all observers in parallel about the new value

  member observable.OnNext value =

   notifyObservers (fun observer -> observer.OnNext value)

 

  /// Notifies all observers in parallel about the error

  /// and finishes all observations

  member observable.OnError error =

    notifyObservers (fun observer -> observer.OnError error)

    observers <- []

 

  /// Notifies all observers in parallel about the completion

  /// and finishes all observations

  member observable.Completed =

    notifyObservers (fun observer -> observer.OnCompleted())

    observers <- [] 

I hope there will be a similar base class in the .NET Framework 4.0 RTM.

Now we are able to use this class and to build our dictionary lookup observable:

let wordsObservable = new Observable.Observable<_>()

 

let findWords prefix =

  if prefix <> "" then

    let prefix’ = prefix.ToUpper()

    for word in data do

      if word.ToUpper().StartsWith(prefix’) then

        wordsObservable.OnNext (prefix,word)

The last step is to create observers and subscribe them to the observables:

// create observers

let clean =

  textChanged

    |> Observable.subscribe (fun _ -> resultsBox.Items.Clear())

 

let searchForWords =

  // Every time the text changes

  // we start our wordsObservable to push words

  textChanged

    |> Observable.subscribe (fun text -> findWords text)

 

let wordFound =

  // subscribe to the "word found"-event

  wordsObservable   

    |> Observable.subscribe

        (fun (_,word) -> resultsBox.Items.Add word |> ignore)

Tags: , , ,

Tuesday, 20. October 2009


IObservable/IObserver – Using the Reactive Framework with F#

Filed under: F# — Steffen Forkmann at 16:50 Uhr

One of the nice new features in .NET 4.0 beta 2 is the IObservable<T>/IObserver<T> support from the Reactive Framework (“Rx Framework” or sometimes “LinqToEvents”). It is a really powerful way to use reactive programming in .NET and especially in F# developed by Erik Meijer and his team.

If you want to see some of the beautiful math behind the Reactive Framework you should definitely watch this Expert to Expert video on Channel 9. You can see Brian Beckman and Erik Meijer showing that IObservable<T> is the mathematical dual of IEnumerable<T>.

What can I do with the Rx Framework?

Consider this small sample (it is taken from Matthew Podwysocki’s blog): We want to get notified whenever a user clicks on our form and moves the mouse within a special area (XPos and YPos smaller than 100px).

First of all we define our observable by merging and filtering .NET events:

open System.Windows.Forms

 

let form = new Form(Visible=true, TopMost=true)

 

/// Creates two observables

///  – left is triggered when the left mouse button is down

         and the mouse is in the area (x < 100 && y < 100)

///  – right is triggered when the right mouse button is down

         and the mouse is in the area (x < 100 && y < 100)

let left,right =

  form.MouseDown

    |> Observable.merge form.MouseMove

    |> Observable.filter (fun args -> 

          args.Button = MouseButtons.Left ||

          args.Button = MouseButtons.Right)

    |> Observable.map (fun args -> args.X, args.Y, args.Button)

    |> Observable.filter (fun (x,y,b) -> x < 100 && y < 100)

    |> Observable.partition (fun (_,_,button) -> button = MouseButtons.Left)

Now it’s easy to subscribe a function to this observable:

let leftSubscription =

  left

    |> Observable.subscribe

         (fun (x,y,_) -> printfn "Left (%d,%d)" x y)

If we want to unsubscribe we only have to dispose the object:

// unsubscribe

leftSubscription.Dispose()

We couldn’t unsubscribe this way with “classic” .NET events. Remember the –= operator in C# doesn’t work with lambda expressions.

Exception handling

We have seen an easy way to subscribe and unsubscribe to complicated observables but what should we do if an error occurs? As far as I know this case is not implemented for F# at the moment, but we can easily add this functionality:

module Observable

 

/// Creates an observer with the given functions

let createObserver next error completed =

    {new System.IObserver<_> with

        member this.OnCompleted() = completed()

        member this.OnError(e) = error e

        member this.OnNext(args) = next args}

 

/// Subscribes an observer with the given functions

///   param1: OnNext        (T -> unit)

///   param2: OnError       (Exception -> unit)

///   param3: OnCompleted   (unit -> unit)

///   param4: observable

let subscribeComplete next error completed (observable:System.IObservable<_>) =

  createObserver next error completed

    |> observable.Subscribe

Now we are able to create a complete IObserver<T> object and register the 3 functions:

let rightSubscription =

  right

    |> Observable.subscribeComplete

         (fun (x,y,_) -> printfn "Right (%d,%d)" x y)

         (fun error   -> printfn "Error: %s" error.Message)

         (fun ()      -> printfn "Ready.")

Tags: , ,