Requêtes HTTP asynchrones avec RxJava

Disons que nous développons un service qui doit interagir avec d’autres composants. Malheureusement, ces composants sont lents et bloquants.

Il peut s’agir d’un service hérité très lent ou d’une API bloquante que nous devons utiliser. Quoi qu’il en soit, nous n’avons aucun contrôle dessus. Dans cet article, nous appellerons deux API. L’un d’eux bloquera pendant deux secondes et un autre pendant cinq secondes.

Nous devons également imprimer les codes d’état des réponses une fois que les deux réponses sont disponibles. Si nous le faisions à l’ancienne, de manière non réactive, nous bloquerions un thread appelant pendant cinq secondes. Tenir le fil pendant cinq secondes n’est pas efficace, n’est-ce pas ?

Prestations de service

J’ai utilisé « httpstat.us » comme service Web. Il s’agit d’un service simple permettant de générer différents codes HTTP pour tester les clients Web. Il est possible de fournir des paramètres supplémentaires, dans ce cas, sleepqui bloque les requêtes HTTP pendant une durée donnée.

Je vais utiliser « httpie » pour tester les deux services.

Le service 1 bloquera pendant cinq secondes et renverra une réponse avec le code d’état 200 :

http://httpstat.us/200?sleep=5000
_____________________________________________

HTTP/1.1 200 OK
Content-Length: 6
Content-Type: text/plain
Date: Tue, 08 Mar 2022 17:05:08 GMT
Request-Context: appId=cid-v1:1e93d241-20e4-4513-bbd7-f452a16a5d69
Server: Kestrel
Set-Cookie: ARRAffinity=e2c17206c539113795daf64bd958d003f2b29b9f62da53617beea05468875ba5;Path=/;HttpOnly;Domain=httpstat.us

200 OK

Le service 2 est identique au précédent sauf qu’il bloque pendant deux secondes au lieu de cinq :

http://httpstat.us/200?sleep=2000
_____________________________________________

HTTP/1.1 200 OK
Content-Length: 6
Content-Type: text/plain
Date: Tue, 08 Mar 2022 17:11:53 GMT
Request-Context: appId=cid-v1:1e93d241-20e4-4513-bbd7-f452a16a5d69
Server: Kestrel
Set-Cookie: ARRAffinity=e2c17206c539113795daf64bd958d003f2b29b9f62da53617beea05468875ba5;Path=/;HttpOnly;Domain=httpstat.us

200 OK

Client Web

Nous avons appris les services. Parlons maintenant du client Web. Dans cette section, j’ai utilisé le client Web Vert.x. C’est un outil asynchrone, facile à utiliser HTTP et HTTP/2 client qui prend également en charge RxJava.

private static Single<Integer> service1(WebClient webClient) {
        return webClient.getAbs("http://httpstat.us/200?sleep=5000")
                .rxSend()
                .doOnSuccess(response -> out.println("[" + Thread.currentThread().getName() + "] service 1: response received"))
                .map(HttpResponse::statusCode);
}

private static Single<Integer> service2(WebClient webClient) {
        return webClient.getAbs("http://httpstat.us/200?sleep=2000")
                .rxSend()
                .doOnSuccess(response -> out.println("[" + Thread.currentThread().getName() + "] service 2 response received"))
                .map(HttpResponse::statusCode);
}

Les deux méthodes sont très similaires. Ils prennent WebClient en tant que paramètre et envoyer des requêtes HTTP renvoyant Single<Integer>. Où l’entier est un code de réponse HTTP. RxJava de retour Single nous assure que le résultat est asynchrone. Le code d’état sera accessible plus tard lorsqu’il sera disponible. Cela nous donne également une évaluation paresseuse, ce qui signifie que les services ne seront invoqués que si un abonnement actif est présent.

Consommer des sources uniques

Il y a deux sources auxquelles nous devrons nous abonner. RxJava a une méthode pratique pour combiner Single sources ensemble. Nous pouvons invoquer la méthode .zipWith sur la première source et fournissez deux paramètres. Le premier est la source avec laquelle compresser et le second est une fonction pour consommer les deux résultats, les traiter et renvoyer autre chose.

Dans ce cas, le type de retour est AbstractMap.SimpleEntry<Integer, Integer>, qui est un tuple simple de deux entiers. Ça a l’air verbeux, n’est-ce pas ? Malheureusement, il n’y a pas de meilleures implémentations de tuples ou de paires dans les bibliothèques Java principales.

Grâce à Java Lambdas, nous pouvons passer le comportement en paramètre :

Single<Integer> service1Code = service1(webClient);
Single<Integer> service2Code = service2(webClient);

Single<AbstractMap.SimpleEntry<Integer, Integer>> tupleSource = 
            service1Code.zipWith(service2Code, (s1, s2) -> new AbstractMap.SimpleEntry<>(s1, s2));

Note: Vous pouvez implémenter votre propre tuple ou paire si un AbstractMap.SimpleEntry se sent trop verbeux.

Tous ensemble

Enfin, nous pouvons assembler tous les morceaux et les paix comme indiqué ci-dessous :

// Vertx instance and web client
Vertx vertx = Vertx.vertx();
WebClient webClient = WebClient.create(vertx);

// single sources. Lazy evaluation, no invocation at this point
Single<Integer> service1Code = service1(webClient);
Single<Integer> service2Code = service2(webClient);

// combine results together and create tuple
Single<AbstractMap.SimpleEntry<Integer, Integer>> tupleSource =
            service1Code.zipWith(service2Code, (s1, s2) -> new AbstractMap.SimpleEntry<>(s1, s2));

// subscribe and invoke services
tupleSource
    .doFinally(countDownLatch::countDown)
    .subscribe(Services::printResult);

Voici les résultats imprimés sur la console après l’exécution du code. Les deux demandes ont été envoyées par le même vertx thread de boucle d’événement. Le programme imprime également des messages indiquant que le thread n’est pas bloqué à chaque seconde. Enfin, il imprime les deux codes d’état comme résultat final. Comme vous pouvez le voir, tout s’est passé sur le même fil :

[vert.x-eventloop-thread-1] is released
[vert.x-eventloop-thread-1] is released
[vert.x-eventloop-thread-1] service 2 response received
[vert.x-eventloop-thread-1] is released
[vert.x-eventloop-thread-1] is released
[vert.x-eventloop-thread-1] is released
[vert.x-eventloop-thread-1] service 1: response received
[vert.x-eventloop-thread-1] Result: service1:200 service2:200

Conclusion

C’est tout ce dans quoi nous nous embarquons pour le moment. J’espère que cet article a été instructif et que vous comprenez maintenant mieux comment envoyer de longues requêtes de blocage de manière asynchrone avec RxJava et Vertx. Si vous êtes intéressé, le code complet utilisé dans cet article est lié ici.