Vous connaissez Rx ? (Reactive Extension)

Ça fait un peu plus d’un an que ça existe, mais encore peu de programmeurs lui ont porté de l’attention.  Pourtant, lorsque vous commencez à l’utilisez, vous en comprenez rapidement les avantages.

En tant qu’architecte et développeur chez nVentive, nous venons justement de terminer un important projet.  Il s’agit de la version Windows Phone 7 et Windows 7 Slate de Flickr dont vous pouvez avoir un aperçu à l’adresse suivante : http://www.flickr.com/windows7.  Sans Reactive Extension, je ne crois pas qu’il aurait été possible de livrer l’application dans les temps tout en ayant une application performante et dont le code doit être le plus maintenable possible.

Qu’est-ce que Reactive Extension ?

Il s’agit d’une nouvelle manière de programmer qui regroupe plusieurs des paradigmes des nouvelles philosophies de programmation :

• Son utilisation est “Fluent” (voir article de Wikipedia : http://en.wikipedia.org/wiki/Fluent_interface).
• Il permet de faire, en C#, une forme de programmation déclarative (aussi appelée programmation fonctionnelle).  En gros, votre code se situe principalement dans les sections d’initialisation des composants en définissant ce qui devra s’exécuter lorsque certaines opérations arriveront.
• Il fait un usage intensif des Generics propres à .NET (rien de nouveau ici), des Methods Extensions, mais, et surtout, de LINQ.
• Il s’agit de l’implémentation des interfaces IObserver<T> et IObservable<T> qui sont apparues dans le namespace “System” de .NET 4.0.  Peu de programmeurs ne semblent les avoir remarqués.

Voyons un exemple rapide de Rx en application :

Observable

// Connexion à l’événement MouseMove

.FromEvent<MouseEventHandler, MouseEventArgs>(

h => h.Invoke, h => this.MouseMove += h, h => this.MouseMove -= h)

// Extraction de la position de la souris

.Select(evt => evt.EventArgs.GetPosition(this))

// Seulement si le X > 200 (c’est une démo, c’est pas obligé d’être logique!)

.Where(position => position.X > 200)

// On veut pas traiter plus de 4 positions à la seconde

.Sample(TimeSpan.FromMilliseconds(250))

// Affichage de la position de la souris dans un Label

.Subscribe(position => txt.Text = position.ToString());

Analysez cette exemple et tentez de voir comment il fonctionne.  Ce que ça fait, c’est que ça va indiquer la position de la souris dans le champ “txt” 4 fois par seconde, mais uniquement lorsque la souris se déplace sur un X supérieur à 200.

Je suppose que vous comprenez maintenant pourquoi Rx (le “petit nom” de Reactive Extension) est souvent décrit comme un LINQ TO THE FUTURE, ou, autrement, comme un LINQ TO EVENTS.

Mais…  on a déjà ça avec les event handlers, non ?

C’est en effet la première réaction que la majorité des programmeurs ont devant Rx.  Même moi je l’ai eue la première fois.

Alors, oui : les events handlers ont déjà ça.  Mais en généralisant les événements vers des IObservable<T>, ça permet d’avoir une manière générique et unifiée de traiter plusieurs sources qui peuvent ne pas être des événements.

De plus, imaginez que vous devez faire le code qui est plus haut à l’aide d’un classique eventhandler : vous allez avoir besoin de définir des variables servant à accumuler des états (“states”), sans compter que le code sera beaucoup moins concis.

Mais il y a un autre aspect non-négligeable avec Reactive Extension : la déconnexion des handlers est automatique lorsque l’on en a plus besoin.  Si, par exemple, je mettais un “Take(10)” avant le .Subscribe() dans mon exemple ci-haut, non seulement ça ferait en sorte qu’il n’y aura pas plus de 10 affichages de la photo, mais tout le event handler se déconnectera automatiquement à ce moment car il n’aura plus d’utilité!  (je vais vous expliquer plus tard comment cette magie s’opère)

Comment ça marche ?

Le fonctionnement de base est tout simple, mais il demande d’un peu oublier la manière dont vous programmez en ce moment.  Je vais essayer de vous l’expliquer en utilisant l’analogie qu’utilisent les créateurs de Rx, eux-mêmes.

On peut d’abord voir Rx comment étant une technologie “Push” (pousser).  Ce à quoi vous êtes probablement habitué, c’est de faire du “Pull” (tirer) : vous faites des applications qui vont chercher des données et attendent leur retour.  Avec Rx, c’est le contraire : vous faites du code qui va réagir à des données reçues qui ne sont pas nécessairement attendues (ou sollicitées).

En programmation impérative, vous écrivez un Pull de la manière suivante :

foreach(string s in source)

{

Console.WriteLine(s.ToString());

}

Voyons ce qui peut arriver lors de cette énumération :

1. Vous pouvez itérer une nouvelle valeur (=Next)
2. Vous pouvez compléter l’énumération (fin de la boucle) (=Completed)
3. Vous pouvez obtenir une exception (supposons qu’un y ait un null dans l’énumération) (=Error)

Eh bien, dans un mode Rx, on va carrément inverser le processus : on va faire en sorte que la source de données « pousse” ses données plutôt que de se les faire “tirer”.  Ainsi, la source deviendra un IObservable<string> et lors de l’utilisation, on va lui fournir un IObserver<string>.  Voyons ces interfaces sans plus attendre :

public interface IObservable<T>

{

IDisposable Subscribe(System.IObserver<T>observer);

}

public interface IObserver<T>

{

void OnCompleted();

void OnError(Exception exception);

void OnNext(T value);

}

Ainsi, la source de données (Observable) peut pousser chacune des possibilités vers l’observer qui lui est passé en paramètre.

Génial, non ?

Et Rx dans tout ça ?

Rx, définit toute une panoplie de method extensions qui permettent de générer des observer ad-hoc (comme le .Subscribe() dans mon exemple qui prend un Lambda et le connecte sur le OnNext() du observer généré) et à ajouter une couche “à la LINQ” par dessus les Observable.  C’est “juste” ça !

Mais là où ça devient vraiment puissant, c’est tous les opérateurs qui existent pour contrôler le flux de données.  Vous avez déjà vu .Sample() plus haut, mais il y en a beaucoup d’autre :

• ObserveOn : permet de rediriger le flux sur un autre dispatcher (autrement dit, un autre thread).  Très pratique pour faire des processus dont seul l’affichage du résultat se produit sur le UI Thread.
• Delay : permet d’ajouter un délais aux données qui arrivent.  Attention cependant, vous ne serez plus sur le thread d’origine après le délais!  Il faut faire un ObserveOn(Scheduler.Dispatcher) pour y revenir.
• Throttle : permet d’obtenir la dernière valeur après une période minimale d’inactivité
• Timeout : Après un certain temps, le OnCompleted est automatiquement généré.
• ForkJoin : Attent que le OnCompleted de plusieurs sources soit déclenché
• Et beaucoup beaucoup d’autres…  Voyez sur http://rxwiki.wikidot.com/101samples

Il ne me reste plus qu’à vous souhaiter de beaucoup vous amuser avec cette nouvelle approche de la programmation.

Une fois que vous aurez bien assimilé le concept, allez écouter ce vidéo du PDC 2010 : http://bit.ly/bfzKCk.  Si vous ne comprenez pas tout du premier coup, dites-vous que c’est normal.