BLOG

Le plugin Grails RxJava ajoute le support de Server-Sent Events, rendant possible dans un contrôleur Grails, l'envoi d'événements du serveur vers un client l'ayant sollicité à l'aide de l'API JavaScript EventSource.
L'API EventSource Server-Sent Events (SSE) est standardisée dans le cadre de HTML5 par le W3C.

Le projet Grails 3.3.0 odelia-gina-sse, que vous pouvez retrouver sur Bitbucket, est une mise en pratique du plugin RxJava et de SSE, pour notifier l'utilisateur de manière visuelle, dans une page web, de l'avancé d'un traitement long s'exécutant côté serveur.
La fin de chaque étape du traitement est signalée par l'envoi d'un événement SSE depuis le contrôleur EventsController, et côté client, la réception de cet événement permet de mettre à jour la timeline affichée dans une page web.

La Timeline utilisée dans l'application provient du code proposé par Abhi Sharma.

Voyons comment cela s'articule en partant du contrôleur Grails EventsController : celui-ci implémente le trait Groovy grails.rx.web.RxController qui apporte la propriété rx par laquelle on va pouvoir émettre des événements SSE, dans un style de programmation réactive.
Notez que l'on a ajouté la dépendance de compilation compile "org.grails.plugins:rxjava" dans le projet Grails (fichier gradle.build).

Voici le code du contrôleur EventsController :

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

package odelia.gina.sse

import rx.Observer
import grails.rx.web.RxController
import grails.converters.JSON


class EventsController implements RxController {

    WorkerService workerService

    def steps = [
            ['Extract', { workerService.doSomeWork() }],
            ['Transform', {workerService.doSomeThing() }],
            ['Load', { workerService.doSomeWork() }],
            ['Analyse', { workerService.doSomeWork() }]
        ]

    def index() {
        [steps: steps.collect { it.head() }]
    }

    def sse() {
        rx.stream { Observer observer ->
            steps.eachWithIndex { step, i ->
                step.last().call()
                observer.onNext(
                        rx.render([step: i, text: "${step.head()}...",
                        last: i == steps.size()-1] as JSON)
                )
            }
            observer.onCompleted()
        }
    }
}

La propriété steps du contrôleur définit sous la forme d'une liste, les différentes étapes liées à la timeline ; chaque élément de cette liste est en fait une paire permettant de spécifier à la fois le nom de l'étape et le code à exécuter sous la forme d'une Closure. Vous pouvez donc modifier cette liste à votre convenance.
Noter que l'on a recours au service WorkerService, dont la référence workerService est obtenue par injection.

Le contrôleur EventsController définit les actions index() et sse().

L'action index() est associée à la vue .gsp views/events/index.gsp, et l'on passe à cette dernière, un modèle de données constitué des noms des étapes, dans le but de construire dynamiquement le code HTML de la timeline.
La vue index.gsp inclut le code JavaScript (fichier assets/javascripts/timeline.js) créant un objet EventSource qui ouvre une connexion vers le serveur avec l'URL passée en argument, soit /events/sse (l'URL relative sse fonctionnerait également), de manière à commencer à recevoir les événements dans la fonction affectée à onmessage :

1
2
3
4

var eventSource = new EventSource("/events/sse");
eventSource.onmessage = function(event) {
	// ...
}

L'objet reçu dans le paramètre event comporte dans son attribut data une chaîne de caratères, construite par le contrôleur Grails, qui sera analysée comme un objet JSON par la fontion JSON.parse() afin de mettre à jour la timeline.

La seconde action sse() du contrôleur est celle qui permet de générer les événements SSE ; pour cela, la méthode steam() appelée sur la propriété rx autorise l'envoi des événement SSE par des appels successifs à observer.onNext(rx.render(...)).
Dans la bouble établie par steps.eachWithIndex, on itère sur chaque étape du traitement en réalisant ceci :

• exécution de la partie du traitement associée à l'étape par step.last().call() ;
• émission de l'événement SSE par l'exécution de observer.onNext(rx.render(...)) qui pousse l'événement vers l'observer.

L'observer est notifié de la fin de l'envoi des événements par l'appel observer.onCompleted().

Notez que les données transmises dans l'instruction rx.render() est une map Groovy convertie en objet JSON, grâce au convertisseur Grails grails.converters.JSON.

Voici un exemple d'objet JSON qui sera reçu dans le code JavaScript, côté client donc :

1

{"step": 0, "text": "Extract...", "last": false}

La dernière paire, de nom last, indique au client s'il s'agit du dernier événement émis, ce qui lui permettra de fermer la connexion, sinon il y a une reconnexion automatique au bout de quelques secondes.

N'hésitez pas à vous reporter à la documentation de RxJava, et à utiliser le projet odelia-gina-sse pour réaliser vos propres expérimentations.
Le site web de Grails propose également le guide Sending Server Sent Events with Grails !