How to call two or more web services or REST in parallel with Project Reactor and join the answers

Question

Hello I would like to know how to call two or more web services or Rest services in pararelo and compose a response of the calls.

I have found some examples on the web using other technologies but I can not get it to work with a reactor

// start task A asynchronously
CompletableFuture<ResponseA> futureA = asyncServiceA.someMethod(someParam);
// start task B asynchronously
CompletableFuture<ResponseB> futureB = asyncServiceB.someMethod(someParam);

CompletableFuture<String> combinedFuture = futureA
        .thenCombine(futureB, (a, b) -> a.toString() + b.toString());

// wait till both A and B complete
String finalValue = combinedFuture.join();

////////////////////////////////////////////////////////////////////////////////

static void Run()
{
    //Follow steps at this link for addding a reference to the necessary .NET library:
    //http://stackoverflow.com/questions/9611316/system-net-http-missing-from-    
    //namespace-using-net-4-5

    //Create an HTTP Client
    var client = new HttpClient();

    //Call first service
    var task1 = client.GetAsync("http://www.cnn.com");

    //Call second service
    var task2 = client.GetAsync("http://www.google.com");

    //Create list of all returned async tasks
    var allTasks = new List<Task<HttpResponseMessage>> { task1, task2 };

    //Wait for all calls to return before proceeding
    Task.WaitAll(allTasks.ToArray());

}

Answer 1

Let's imagine you need to hit 2 services, so you nee 2 base WebClient (each is configured with the correct base URL and eg. an authentication scheme):

@Bean
public WebClient serviceAClient(String authToken) {
    return WebClient.builder()
            .baseUrl("http://serviceA.com/api/v2/")
            .defaultHeader(HttpHeaders.AUTHORIZATION, "Basic " + authToken)
            .defaultHeader(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)
            .build();
}

@Bean
public WebClient serviceBClient(String authToken): WebClient {
    return WebClient.builder()
            .baseUrl("https://api.serviceB.com/")
            .defaultHeader(HttpHeaders.AUTHORIZATION, "token " + authToken)
            .defaultHeader(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)
            .build();
}

From there, let's assume you get these 2 webclients injected in your controller (as qualified beans). Here's the code to make a joint call to both using Reactor:

Mono<ResponseA> respA = webclientA.get()
                                 .uri("/sub/path/" + foo)
                                 .retrieve()
                                 .bodyToMono(ResponseA.class);
Mono<ResponseB> respB = webclientB.get()
                                 .uri("/path/for/b")
                                 .retrieve()
                                 .bodyToMono(ResponseB.class);
Mono<String> join = respA.zipWith(respB, (a, b) -> a.toString + b.toString);
return join;

Note the zip function could produce something more meaningful like a business object out of the 2 responses. The resulting Mono<String> only triggers the 2 requests if something subscribes to it (in the case of Spring WebFlux, the framework will do that if you return it from a controller method).

Answer 2

If you´re using Spring reactor what you need is the operator Zip, to run your process and zip them once are finished.

/**
 * Zip operator execute the N number of Flux independently, and once all them are finished, results
 * are combined in TupleN object.
 */
@Test
public void zip() {
    Flux<String> flux1 = Flux.just("hello ");
    Flux<String> flux2 = Flux.just("reactive");
    Flux<String> flux3 = Flux.just(" world");
    Flux.zip(flux1, flux2, flux3)
            .map(tuple3 -> tuple3.getT1().concat(tuple3.getT2()).concat(tuple3.getT3()))
            .map(String::toUpperCase)
            .subscribe(value -> System.out.println("zip result:" + value));
}

You can see more about reactive technology here https://github.com/politrons/reactive